import asyncio
import aiohttp
import re
import uuid
import json
import math
from typing import Dict, Any, List, Tuple
from datetime import datetime
from sqlalchemy.orm import Session
from app.crud.ai_workflow import ai_workflow
from app.models.ai_workflow import TaskStatus
from app.services.ai.coze_service import coze_service
from app.services.ai.key_pool import get_zeakai_key_pool
from app.services.tencent.cos_service import TencentCOSService
from app.services.image.processor import ImageProcessor
from app.crud.image import image_crud

class WorkflowProcessor:
    def __init__(self, db: Session):
        self.db = db
        # 延迟到协程内创建，避免跨事件循环
        self._save_lock = None
        # 不在此处创建/持有跨循环的 key pool
        self._key_pool = None
        # 共享的HTTP会话，避免频繁创建连接器
        self._http_session = None
        self._connector = None
    
    def __del__(self):
        """析构函数，确保HTTP会话被正确关闭"""
        if self._http_session and not self._http_session.closed:
            # 在析构时尝试关闭会话
            try:
                import asyncio
                loop = asyncio.get_event_loop()
                if loop.is_running():
                    loop.create_task(self._close_http_session())
                else:
                    loop.run_until_complete(self._close_http_session())
            except Exception:
                # 如果无法异步关闭，至少尝试同步关闭
                pass
    
    async def __aenter__(self):
        """异步上下文管理器入口"""
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """异步上下文管理器出口，确保资源清理"""
        await self._close_http_session()
    
    def _get_http_session(self):
        """获取HTTP会话上下文管理器，确保会话正确关闭"""
        connector = aiohttp.TCPConnector(
            limit=20,  # 增加连接池大小
            limit_per_host=10,  # 增加每个主机的连接数
            keepalive_timeout=60,  # 增加保活时间
            enable_cleanup_closed=True,
            ssl=False,  # 禁用SSL验证以避免SSL握手问题
            ttl_dns_cache=300,  # DNS缓存5分钟
            use_dns_cache=True
        )
        
        # 返回会话上下文管理器
        return aiohttp.ClientSession(
            connector=connector,
            timeout=aiohttp.ClientTimeout(total=60),  # 默认60秒超时
            headers={"User-Agent": "WorkflowProcessor/1.0"}
        )
    
    async def _close_http_session(self):
        """关闭HTTP会话"""
        if self._http_session and not self._http_session.closed:
            await self._http_session.close()
        if self._connector:
            await self._connector.close()
        self._http_session = None
        self._connector = None
    
    async def execute_full_pipeline(self, workflow_id: int, image_path: str, user_prompt: str):
        """执行完整的AI处理管道"""
        try:
            # 添加状态检查函数
            def check_if_cancelled():
                workflow = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
                if workflow and workflow.status == TaskStatus.CANCELLED:
                    raise Exception("任务已被取消")
            
            # 检查当前状态，如果已经是RUNNING则不重复设置
            workflow = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
            if workflow and workflow.status != TaskStatus.RUNNING:
                # 只有当状态不是RUNNING时才更新为处理中
                ai_workflow.update_workflow_status(
                    db=self.db, 
                    workflow_id=workflow_id, 
                    status=TaskStatus.RUNNING,
                    current_step="开始处理",
                    progress=0
                )
            
            # 在每个关键步骤前检查是否被取消
            check_if_cancelled()
            
            # 初始化result_data，将所有步骤结果设为默认安全值（避免 None 持久化为 null）
            result_data = {
                "optimized_prompt": "",
                "mj_image_urls": [],
                "split_images": [],
                "flux_variations": [],
                "product_info": [],
                "detail_pages": []
            }
            
            # 步骤1: 智能体反推修改提示词
            check_if_cancelled()  # 检查取消状态
            ai_workflow.update_workflow_status(
                db=self.db, 
                workflow_id=workflow_id,
                status=TaskStatus.RUNNING,
                current_step="智能体优化提示词", 
                progress=15
            )
            
            optimized_prompt = await self.step1_optimize_prompt(image_path, user_prompt, workflow_id)
            result_data["optimized_prompt"] = optimized_prompt
            
            # ✅ 立即保存第一步的结果
            ai_workflow.update_workflow_result(
                db=self.db, workflow_id=workflow_id,
                result_data=result_data
            )
            
            # 步骤2: MJ文生图
            check_if_cancelled()  # 检查取消状态
            ai_workflow.update_workflow_status(
                db=self.db, 
                workflow_id=workflow_id,
                status=TaskStatus.RUNNING,
                current_step="MJ文生图", 
                progress=30
            )
            
            mj_image_urls = await self.step2_mj_generation(optimized_prompt, workflow_id)
            result_data["mj_image_urls"] = mj_image_urls
            
            # 保存第二步的结果
            ai_workflow.update_workflow_result(
                db=self.db, workflow_id=workflow_id,
                result_data=result_data
            )
            
            # 步骤3: MJ图1拆4 - 处理所有MJ生成的图片
            check_if_cancelled()  # 检查取消状态
            ai_workflow.update_workflow_status(
                db=self.db, 
                workflow_id=workflow_id,
                status=TaskStatus.RUNNING,
                current_step="MJ图片拆分", 
                progress=45
            )
            
            all_split_images = []
            # 过滤掉None值和无效的URL
            valid_mj_image_urls = [url for url in mj_image_urls if url and isinstance(url, str) and url.strip()]
            print(f"🔍 有效的MJ图片URL数量: {len(valid_mj_image_urls)}/{len(mj_image_urls)}")
            
            for i, mj_image_url in enumerate(valid_mj_image_urls):
                print(f"🔍 开始拆分第{i+1}张MJ图片: {mj_image_url}")
                try:
                    split_images = await self.step3_mj_split(mj_image_url, workflow_id)
                    if split_images:
                        all_split_images.extend(split_images)
                        print(f"✅ 第{i+1}张MJ图片拆分成功，获得{len(split_images)}张图片")
                    else:
                        print(f"❌ 第{i+1}张MJ图片拆分失败")
                except Exception as e:
                    print(f"❌ 第{i+1}张MJ图片拆分异常: {str(e)}")
                    continue
            
            result_data["split_images"] = all_split_images
            
            # 保存第三步的结果
            ai_workflow.update_workflow_result(
                db=self.db, workflow_id=workflow_id,
                result_data=result_data
            )
            
            if not all_split_images:
                raise Exception("所有MJ图片拆分都失败了")
            
            print(f"🔍 总共拆分得到 {len(all_split_images)} 张图片")
            
            # 步骤4: Flux裂变
            check_if_cancelled()  # 检查取消状态
            ai_workflow.update_workflow_status(
                db=self.db, 
                workflow_id=workflow_id,
                status=TaskStatus.RUNNING,
                current_step="Flux图片裂变", 
                progress=60
            )
            
            flux_variations = await self.step4_flux_variation(all_split_images, workflow_id)
            result_data["flux_variations"] = flux_variations
            
            # 保存第四步的结果
            ai_workflow.update_workflow_result(
                db=self.db, workflow_id=workflow_id,
                result_data=result_data
            )
            
            # 步骤5: 智能体生成标题/产品简介/标签
            check_if_cancelled()  # 检查取消状态
            ai_workflow.update_workflow_status(
                db=self.db, 
                workflow_id=workflow_id,
                status=TaskStatus.RUNNING,
                current_step="生成产品信息", 
                progress=75
            )
            
            product_info = await self.step5_generate_product_info(optimized_prompt, workflow_id)
            result_data["product_info"] = product_info
            
            # 保存第五步的结果
            ai_workflow.update_workflow_result(
                db=self.db, workflow_id=workflow_id,
                result_data=result_data
            )
            
            # 步骤6: GPT4o生成详情页
            check_if_cancelled()  # 检查取消状态
            ai_workflow.update_workflow_status(
                db=self.db, 
                workflow_id=workflow_id,
                status=TaskStatus.RUNNING,
                current_step="生成详情页", 
                progress=90
            )
            
            detail_pages = await self.step6_generate_detail_page(product_info, flux_variations, workflow_id)
            result_data["detail_pages"] = detail_pages
            
            # 完成
            ai_workflow.update_workflow_status(
                db=self.db, workflow_id=workflow_id,
                status=TaskStatus.COMPLETED,
                current_step="处理完成", progress=100
            )
            
            # 最终保存完整结果
            ai_workflow.update_workflow_result(
                db=self.db, workflow_id=workflow_id,
                result_data=result_data
            )
            
        except Exception as e:
            # 如果是取消异常，更新状态为已取消
            if "任务已被取消" in str(e):
                ai_workflow.update_workflow_status(
                    db=self.db, workflow_id=workflow_id,
                    status=TaskStatus.CANCELLED,
                    error_message="任务已被用户取消"
                )
                print(f"✅ 任务 {workflow_id} 已成功取消")
            else:
                ai_workflow.update_workflow_status(
                    db=self.db, workflow_id=workflow_id,
                    status=TaskStatus.FAILED,
                    error_message=str(e)
                )
            raise
        finally:
            # 确保关闭HTTP会话
            try:
                await self._close_http_session()
            except Exception as close_error:
                print(f"⚠️ 关闭HTTP会话时出错: {str(close_error)}")
    
    async def step1_optimize_prompt(self, image_path: str, user_prompt: str, workflow_id: int) -> str:
        """步骤1: 智能体优化提示词"""
        workflow = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
        # 新增：参数校验，避免 None 触发 AttributeError
        if not image_path:
            raise Exception("未提供图片路径或URL：请确保 input_data 中包含 image_url 或 image_urls，并且至少有一项不为空")
        # 确保image_path是完整的URL
        if not image_path.startswith('http'):
            image_url = f"https://your-domain.com/{image_path}"
        else:
            image_url = image_path
            
        try:
            print(f"🔍 开始调用反推提示词接口...")
            print(f"🔍 图片URL: {image_url}")
            print(f"🔍 用户ID: {workflow.user_id}")
            print(f"🔍 工作流ID: {workflow_id}")
            
            # 调用异步方法
            success, result = await coze_service.process_image_optimization_async(
                user_id=workflow.user_id,
                image_url=image_url,
                db=self.db,
                workflow_id=workflow_id
            )
            
            print(f"🔍 Coze接口调用结果:")
            print(f"🔍 成功状态: {success}")
            print(f"🔍 返回内容类型: {type(result)}")
            print(f"🔍 返回内容长度: {len(str(result))}")
            
            if success:
                print(f"✅ 反推提示词成功")
                return result  # 直接返回接口原始数据
            else:
                raise Exception(f"图片反推优化失败: {result}")
                
        except Exception as e:
            print(f"❌ step1_optimize_prompt 异常: {str(e)}")
            raise Exception(f"图片反推优化失败: {str(e)}")

    async def step2_mj_generation(self, prompt: str, workflow_id: int, image_url: str = None, resume_from_existing: bool = False) -> list:
        """步骤2: MJ文生图 - 处理所有提示词组，支持断点续传和增量保存"""
        try:
            # 添加必要的导入
            from app.services.tencent.cos_service import TencentCOSService
            from app.crud.image import image_crud
            import uuid
            import json
            
            print(f"🔍 step2_mj_generation 开始...")
            print(f"🔍 断点续传模式: {resume_from_existing}")
            print(f"🔍 接收到的完整提示词:")
            print(f"{'='*50}")
            print(prompt)
            print(f"{'='*50}")
            
            # 获取工作流信息，从input_data中获取图片路径
            workflow = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
            
            # 检查是否有已存在的MJ图片（断点续传）
            existing_images = []
            if resume_from_existing:
                result_data = workflow.result_data or {}
                existing_mj_images = result_data.get('mj_image_urls') or result_data.get('mj_image_url')
                if existing_mj_images:
                    if isinstance(existing_mj_images, list):
                        existing_images = existing_mj_images
                    else:
                        existing_images = [existing_mj_images]
                print(f"🔍 发现已存在的MJ图片: {len(existing_images)}张")
                for i, img in enumerate(existing_images):
                    print(f"🔍 已有第{i+1}张: {img}")
            
            # 优先使用传入的image_url参数，如果没有则从input_data中获取
            if not image_url:
                input_data = workflow.input_data or {}
                image_url = input_data.get('image_path') or input_data.get('image_url')
            
            if not image_url:
                raise Exception("未找到图片路径")
            
            print(f"🔍 原始图片URL: {image_url}")
            
            # 从Coze返回的内容中提取所有提示词组
            lines = prompt.split('\n')
            prompts_list = []
            
            print(f"🔍 分割后的行数: {len(lines)}")
            # 只输出前5行用于调试
            for i, line in enumerate(lines[:5]):
                print(f"🔍 第{i+1}行: {repr(line)}")
            
            # 更灵活的提示词提取逻辑
            for line in lines:
                line = line.strip()
                
                # 跳过空行和标题行
                if not line or '生成的英文提示词内容' in line:
                    continue
                
                # 方法1: 查找包含"反推英文提示词内容"的行
                if '反推英文提示词内容' in line:
                    extracted_prompt = line.split('反推英文提示词内容')[-1].strip()
                    if extracted_prompt.startswith('-'):
                        extracted_prompt = extracted_prompt.lstrip('- ').strip()
                    if extracted_prompt and len(extracted_prompt) > 10:  # 确保不是空字符串且有实际内容
                        print(f"🔍 方法1提取到: {extracted_prompt[:50]}...")
                        prompts_list.append(extracted_prompt)
                        continue
                
                # 方法2: 查找以"- "开头的行（但排除标题行）
                if line.startswith('- ') and len(line) > 10:
                    extracted_prompt = line.lstrip('- ').strip()
                    # 排除明显的标题或说明文字
                    if not any(keyword in extracted_prompt.lower() for keyword in ['生成', '提示词', '内容', '英文']):
                        print(f"🔍 方法2提取到: {extracted_prompt[:50]}...")
                        prompts_list.append(extracted_prompt)
                        continue
                
                # 方法3: 查找数字开头的行（如"1. "、"2. "等）
                if line and line[0].isdigit() and ('. ' in line or ') ' in line):
                    # 提取数字后的内容
                    if '. ' in line:
                        extracted_prompt = line.split('. ', 1)[1].strip()
                    elif ') ' in line:
                        extracted_prompt = line.split(') ', 1)[1].strip()
                    
                    if extracted_prompt and len(extracted_prompt) > 10:
                        print(f"🔍 方法3提取到: {extracted_prompt[:50]}...")
                        prompts_list.append(extracted_prompt)
                        continue
                
                # 方法4: 如果行内容较长且不包含中文，可能是英文提示词
                if len(line) > 20 and not any('\u4e00' <= char <= '\u9fff' for char in line):
                    # 检查是否像英文提示词（包含常见的英文单词）
                    english_keywords = ['a', 'an', 'the', 'and', 'or', 'with', 'in', 'on', 'at', 'of', 'for']
                    if any(keyword in line.lower().split() for keyword in english_keywords):
                        print(f"🔍 方法4提取到: {line[:50]}...")
                        prompts_list.append(line)
            
            print(f"🔍 提取到的提示词组数量: {len(prompts_list)}")
            # 调试：输出前3个提示词以检查是否重复
            for i, p in enumerate(prompts_list[:3]):
                print(f"🔍 调试-第{i+1}组提示词: {p[:100]}...")
            
            # 如果还是没有提取到，尝试直接使用原始内容
            if not prompts_list:
                print(f"🔍 使用备用方案：直接使用原始提示词内容")
                # 移除明显的标题行，保留可能的提示词内容
                for line in lines:
                    line = line.strip()
                    if line and len(line) > 10 and '生成的英文提示词内容' not in line:
                        prompts_list.append(line)
                
                # 如果还是没有，至少保证有一个提示词
                if not prompts_list:
                    prompts_list = [prompt.strip()]  # 使用整个内容作为一个提示词
            
            if not prompts_list:
                raise Exception("未能提取到有效的提示词")
            
            # 计算需要处理的提示词（跳过已完成的）
            # 🔧 修复：从工作流的expected_outputs中获取实际需要的MJ图片数量
            expected_outputs = workflow.expected_outputs or {}
            total_needed = expected_outputs.get('mj_images', 10)  # 默认10张图
            already_completed = len(existing_images)
            remaining_needed = total_needed - already_completed
            
            print(f"🔍 总共需要: {total_needed}张图")
            print(f"🔍 已完成: {already_completed}张图")
            print(f"🔍 还需要: {remaining_needed}张图")
            
            if remaining_needed <= 0:
                print(f"✅ MJ图片已全部完成，返回已有图片")
                return existing_images
            
            # 确定要处理的提示词范围
            # 🔧 修复：如果提示词数量不足，先尝试重新获取新的提示词，再循环使用现有提示词
            if len(prompts_list) < total_needed:
                print(f"⚠️ 提示词数量不足({len(prompts_list)})，需要{total_needed}张图片")
                
                # 尝试重新获取新的提示词
                try:
                    print(f"🔄 尝试重新获取新的提示词...")
                    from app.services.ai.coze_service import CozeAPIService
                    coze_service = CozeAPIService()
                    
                    # 重新调用反推提示词接口
                    success, new_prompt_result = await coze_service.process_image_optimization_async(
                        user_id=workflow.user_id,
                        image_url=image_url,
                        db=self.db,
                        workflow_id=workflow_id
                    )
                    
                    if success and new_prompt_result:
                        print(f"✅ 成功获取新的提示词，长度: {len(str(new_prompt_result))}")
                        
                        # 解析新获取的提示词
                        new_lines = str(new_prompt_result).split('\n')
                        new_prompts_list = []
                        
                        for line in new_lines:
                            line = line.strip()
                            if not line or '生成的英文提示词内容' in line:
                                continue
                            
                            # 使用相同的提取逻辑
                            if '反推英文提示词内容' in line:
                                extracted_prompt = line.split('反推英文提示词内容')[-1].strip()
                                if extracted_prompt.startswith('-'):
                                    extracted_prompt = extracted_prompt.lstrip('- ').strip()
                                if extracted_prompt and len(extracted_prompt) > 10:
                                    new_prompts_list.append(extracted_prompt)
                                    continue
                            
                            if line.startswith('- ') and len(line) > 10:
                                extracted_prompt = line.lstrip('- ').strip()
                                if not any(keyword in extracted_prompt.lower() for keyword in ['生成', '提示词', '内容', '英文']):
                                    new_prompts_list.append(extracted_prompt)
                                    continue
                            
                            if line and line[0].isdigit() and ('. ' in line or ') ' in line):
                                if '. ' in line:
                                    extracted_prompt = line.split('. ', 1)[1].strip()
                                else:
                                    extracted_prompt = line.split(') ', 1)[1].strip()
                                if extracted_prompt and len(extracted_prompt) > 10:
                                    new_prompts_list.append(extracted_prompt)
                                    continue
                            
                            if len(line) > 20 and not any('\u4e00' <= char <= '\u9fff' for char in line):
                                english_keywords = ['a', 'an', 'the', 'and', 'or', 'with', 'in', 'on', 'at', 'of', 'for']
                                if any(keyword in line.lower().split() for keyword in english_keywords):
                                    new_prompts_list.append(line)
                        
                        if new_prompts_list:
                            print(f"🔍 从新请求中提取到 {len(new_prompts_list)} 个新提示词")
                            # 合并新旧提示词，去重
                            combined_prompts = prompts_list.copy()
                            for new_prompt in new_prompts_list:
                                if new_prompt not in combined_prompts:
                                    combined_prompts.append(new_prompt)
                            prompts_list = combined_prompts
                            print(f"🔍 合并后的提示词数量: {len(prompts_list)}")
                        else:
                            print(f"⚠️ 新请求未能提取到有效提示词")
                    else:
                        print(f"⚠️ 重新获取提示词失败: {new_prompt_result}")
                        
                except Exception as e:
                    print(f"⚠️ 重新获取提示词异常: {str(e)}")
                
                # 如果提示词仍然不足，循环使用现有提示词
                if len(prompts_list) < total_needed:
                    print(f"🔄 提示词仍不足，将循环使用现有提示词")
                    original_prompts = prompts_list.copy()  # 保存原始提示词
                    # 循环使用提示词直到满足需求
                    while len(prompts_list) < total_needed:
                        # 计算还需要多少个提示词
                        needed = total_needed - len(prompts_list)
                        # 从原始提示词中循环取用
                        for i in range(min(needed, len(original_prompts))):
                            prompts_list.append(original_prompts[i])
                    print(f"🔍 最终提示词数量: {len(prompts_list)}")
                    # 调试：输出扩展后的前几个提示词
                    for i, p in enumerate(prompts_list[:5]):
                        print(f"🔍 最终第{i+1}组: {p[:50]}...")
            
            prompts_to_process = prompts_list[already_completed:already_completed + remaining_needed]
            print(f"🔍 本次需要处理的提示词数量: {len(prompts_to_process)}")
            
            # 步骤1: 先下载图片并转换为base64（添加重试机制）
            print(f"🔍 开始下载图片...")
            
            # 添加重试机制处理连接错误
            max_download_retries = 3
            download_retry_count = 0
            image_data = None
            
            while download_retry_count < max_download_retries:
                try:
                    # 使用HTTP会话上下文管理器
                    async with self._get_http_session() as session:
                        async with session.get(image_url) as response:
                            if response.status != 200:
                                raise Exception(f"下载图片失败: {response.status}")
                            image_data = await response.read()
                            break  # 下载成功，跳出重试循环
                            
                except Exception as e:
                    download_retry_count += 1
                    error_msg = str(e)
                    print(f"❌ 图片下载失败 (尝试 {download_retry_count}/{max_download_retries}): {error_msg}")
                    
                    # 如果是连接相关错误，等待后重试
                    if any(keyword in error_msg.lower() for keyword in ['connector is closed', 'connection', 'timeout', 'ssl']):
                        if download_retry_count < max_download_retries:
                            wait_time = download_retry_count * 2  # 递增等待时间
                            print(f"🔄 检测到连接错误，等待 {wait_time} 秒后重试...")
                            await asyncio.sleep(wait_time)
                            continue
                    
                    # 如果达到最大重试次数或非连接错误，抛出异常
                    if download_retry_count >= max_download_retries:
                        raise Exception(f"图片下载失败: {error_msg}")
                    else:
                        raise
            
            if not image_data:
                raise Exception("图片下载失败: 未获取到有效的图片数据")
                    
            print(f"🔍 图片下载成功，大小: {len(image_data)} bytes")
            
            # 转换为base64
            import base64
            base64_image = base64.b64encode(image_data).decode('utf-8')
            base64_with_prefix = f"data:image/png;base64,{base64_image}"
            
            print(f"🔍 base64转换完成，长度: {len(base64_with_prefix)}")
            
            # 步骤2: 上传图片到MJ平台（添加重试机制）
            upload_url = "https://zeakai-api.api4midjourney.com/mj/submit/upload-discord-images"
            # 从 KeyPool 获取一个 token，仅用于上传这一步
            key_pool = get_zeakai_key_pool()
            
            # 添加重试机制处理连接错误
            max_upload_retries = 3
            upload_retry_count = 0
            mj_image_url = None
            
            while upload_retry_count < max_upload_retries:
                try:
                    async with key_pool.acquire() as upload_token:
                        upload_headers = {
                            "Authorization": upload_token,   # 注意：MJ 接口原来就是裸 token，不是 Bearer
                            "Content-Type": "application/json"
                        }
                        upload_data = {
                            "base64Array": [base64_with_prefix]
                        }
                        
                        print(f"🔍 MJ图片上传参数 (尝试 {upload_retry_count + 1}/{max_upload_retries}):")
                        print(f"🔍 URL: {upload_url}")
                        
                        # 创建新的HTTP会话
                        async with self._get_http_session() as session:
                            async with session.post(upload_url, json=upload_data, headers=upload_headers) as response:
                                if response.status != 200:
                                    error_text = await response.text()
                                    raise Exception(f"MJ图片上传失败: {response.status}, {error_text}")
                                
                                upload_result = await response.json()
                                print(f"🔍 MJ图片上传响应: {upload_result}")
                                
                                # 从上传响应中获取图片链接
                                if 'result' in upload_result and upload_result['result']:
                                    mj_image_url = upload_result['result'][0]
                                elif 'imageUrls' in upload_result and upload_result['imageUrls']:
                                    mj_image_url = upload_result['imageUrls'][0]
                                elif 'data' in upload_result and 'imageUrls' in upload_result['data']:
                                    mj_image_url = upload_result['data']['imageUrls'][0]
                                else:
                                    error_msg = f"未获取到MJ图片链接: {upload_result}"
                                    raise Exception(error_msg)
                                
                                # 上传成功，跳出重试循环
                                break
                                
                except Exception as e:
                    upload_retry_count += 1
                    error_msg = str(e)
                    print(f"❌ MJ图片上传失败 (尝试 {upload_retry_count}/{max_upload_retries}): {error_msg}")
                    
                    # 如果是连接相关错误，等待后重试
                    if any(keyword in error_msg.lower() for keyword in ['connector is closed', 'connection', 'timeout', 'ssl']):
                        if upload_retry_count < max_upload_retries:
                            wait_time = upload_retry_count * 2  # 递增等待时间
                            print(f"🔄 检测到连接错误，等待 {wait_time} 秒后重试...")
                            await asyncio.sleep(wait_time)
                            continue
                    
                    # 如果达到最大重试次数或非连接错误，抛出异常
                    if upload_retry_count >= max_upload_retries:
                        raise Exception(f"图片上传失败: {error_msg}")
                    else:
                        raise
            
            if not mj_image_url:
                raise Exception("图片上传失败: 未获取到有效的图片链接")
            
            # 定义后续 MJ imagine 请求使用的 headers（沿用原有固定方式）
            headers = {
                "Authorization": "6a375dc8-cddb-4276-9cae-e67247fee125",
                "Content-Type": "application/json"
            }
            
            print(f"🔍 获取到MJ图片链接: {mj_image_url}")
            
            # 步骤3: 逐一处理需要的提示词组（从断点开始）- 增量保存
            imagine_url = "https://zeakai-api.api4midjourney.com/mj/submit/imagine"
            new_generated_images = []
            
            # 将串行 for 循环改为并发任务
            async def process_one(actual_index: int, current_prompt: str):
                try:
                    # 导入ai_workflow模块
                    from app.crud.ai_workflow import ai_workflow
                    
                    # 任务取消检查（与原逻辑一致）
                    workflow = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
                    if workflow and workflow.status == TaskStatus.CANCELLED:
                        print(f"🚫 任务已被取消，停止当前 prompt")
                        return

                    # 构建最终 prompt
                    mj_prompt = f"{mj_image_url} {current_prompt}"
                    request_data = {
                        "base64Array": [],
                        "botType": "MID_JOURNEY",
                        "instanceId": "",
                        "prompt": mj_prompt
                    }
                    print(f"🔍 第{actual_index}张图：提示词 {current_prompt} 图片url {mj_image_url}")

                    # 添加排队上限重试机制
                    max_queue_retries = 5  # 最多重试5次排队上限错误
                    queue_retry_count = 0
                    
                    while queue_retry_count < max_queue_retries:
                        # 每个任务独立从池子取一个 token，用于 imagine + 轮询阶段
                        from app.services.ai.key_pool import get_zeakai_key_pool
                        token_pool = get_zeakai_key_pool()
                        async with token_pool.acquire() as token:
                            headers = {
                                "Authorization": token,  # MJ 接口使用裸 token 头
                                "Content-Type": "application/json"
                            }
                            
                            # 提交 imagine - 改进重试机制
                            task_id = None
                            max_imagine_retries = 3
                            for imagine_retry in range(max_imagine_retries):
                                try:
                                    # 创建新的HTTP会话，设置30秒超时
                                    timeout = aiohttp.ClientTimeout(total=30)
                                    async with self._get_http_session() as session:
                                        async with session.post(imagine_url, headers=headers, json=request_data, timeout=timeout) as resp:
                                            if resp.status != 200:
                                                error_text = await resp.text()
                                                raise Exception(f"第{actual_index}组 - imagine 提交失败: {resp.status}, {error_text}")
                                            result = await resp.json()
                                            task_id = result.get("result") or result.get("taskId") or result.get("data", {}).get("taskId")
                                            if not task_id:
                                                raise Exception(f"第{actual_index}组 - 未获取到 taskId: {result}")
                                            print(f"🔍 第{actual_index}组 - imagine 提交成功，task_id={task_id}")
                                            break
                                except Exception as e:
                                    error_msg = str(e)
                                    print(f"❌ 第{actual_index}组 - imagine 提交失败 (尝试 {imagine_retry + 1}/{max_imagine_retries}): {error_msg}")
                                    
                                    # 检查是否为连接相关错误
                                    is_connection_error = any(keyword in error_msg.lower() for keyword in [
                                        'connector is closed', 'connection', 'timeout', 'ssl', 
                                        'operation timed out', 'cannot connect to host'
                                    ])
                                    
                                    if is_connection_error and imagine_retry < max_imagine_retries - 1:
                                        wait_time = (imagine_retry + 1) * 2  # 递增等待时间
                                        print(f"🔄 第{actual_index}组 - 检测到连接错误，等待 {wait_time} 秒后重试...")
                                        await asyncio.sleep(wait_time)
                                        continue
                                    
                                    # 如果不是连接错误或已达到最大重试次数，抛出异常
                                    if imagine_retry >= max_imagine_retries - 1:
                                        raise Exception(f"第{actual_index}组 - imagine 提交失败: {error_msg}")
                                    else:
                                        raise
                            
                            if not task_id:
                                raise Exception(f"第{actual_index}组 - imagine 提交失败，已重试 {max_imagine_retries} 次")

                            # 轮询任务状态（使用同一 token）- 添加重试机制
                            status_url = f"https://zeakai-api.api4midjourney.com/mj/task/{task_id}/fetch"
                            final_image_url = None
                            queue_limit_hit = False  # 标记是否遇到排队上限
                            for attempt in range(30):
                                # 取消检查
                                workflow = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
                                if workflow and workflow.status == TaskStatus.CANCELLED:
                                    raise Exception("任务已被取消")

                                # 状态轮询重试机制
                                max_poll_retries = 3
                                poll_success = False
                                for poll_retry in range(max_poll_retries):
                                    try:
                                        # 使用HTTP会话上下文管理器，设置20秒超时
                                        timeout = aiohttp.ClientTimeout(total=20)
                                        async with self._get_http_session() as session:
                                            async with session.get(status_url, headers=headers, timeout=timeout) as status_response:
                                                if status_response.status == 200:
                                                    status_result = await status_response.json()
                                                    print(f"🔍 第{actual_index}组 - 轮询第{attempt+1}次，状态: {status_result.get('status','unknown')}")
                                                    if status_result.get('status') == 'SUCCESS':
                                                        final_image_url = (status_result.get('imageUrl') or "").strip()
                                                        if not final_image_url:
                                                            raise Exception(f"第{actual_index}组 - 成功但未获取到 imageUrl: {status_result}")
                                                        poll_success = True
                                                        break
                                                    elif status_result.get('status') in ['FAILURE','FAILED']:
                                                        fail_reason = status_result.get('failReason') or status_result.get('error') or '未知错误'
                                                        # 检查是否为排队上限错误，如果是则重新提交任务而不是直接失败
                                                        if '排队中任务已达上限' in fail_reason or '请等待排队任务完成后重新生成' in fail_reason:
                                                            print(f"🔄 第{actual_index}组 - 检测到排队上限，将重新提交任务: {fail_reason}")
                                                            # 设置标志并跳出轮询循环
                                                            queue_limit_hit = True
                                                            poll_success = True
                                                            break
                                                        else:
                                                            raise Exception(f"第{actual_index}组 - 失败: {fail_reason}")
                                                    # 对于其他状态（如SUBMITTED, IN_PROGRESS等），标记为成功获取状态，继续轮询
                                                    poll_success = True
                                                    break
                                                else:
                                                    print(f"🔄 第{actual_index}组 - 状态查询返回 {status_response.status}，重试 {poll_retry + 1}/{max_poll_retries}")
                                    except Exception as e:
                                        error_msg = str(e)
                                        print(f"❌ 第{actual_index}组 - 状态轮询失败 (尝试 {poll_retry + 1}/{max_poll_retries}): {error_msg}")
                                        
                                        # 检查是否为连接相关错误
                                        is_connection_error = any(keyword in error_msg.lower() for keyword in [
                                            'connector is closed', 'connection', 'timeout', 'ssl',
                                            'operation timed out', 'cannot connect to host'
                                        ])
                                        
                                        if is_connection_error and poll_retry < max_poll_retries - 1:
                                            wait_time = (poll_retry + 1) * 2  # 递增等待时间
                                            print(f"🔄 第{actual_index}组 - 检测到连接错误，等待 {wait_time} 秒后重试...")
                                            await asyncio.sleep(wait_time)
                                            continue
                                        raise e
                                
                                if final_image_url:
                                    break
                                if queue_limit_hit:
                                    # 遇到排队上限，跳出轮询循环
                                    break
                                if not poll_success:
                                    print(f"⚠️ 第{actual_index}组 - 状态轮询失败，等待后继续")
                                await asyncio.sleep(10)

                            # 检查是否因为排队上限而退出
                            if queue_limit_hit:
                                queue_retry_count += 1
                                if queue_retry_count < max_queue_retries:
                                    print(f"🔄 第{actual_index}组 - 排队上限重试 {queue_retry_count}/{max_queue_retries}，等待30秒后重新提交")
                                    await asyncio.sleep(30)  # 等待30秒后重试
                                    break  # 跳出token会话，重新开始
                                else:
                                    raise Exception(f"第{actual_index}组 - 排队上限重试次数已达上限({max_queue_retries}次)")
                            
                            if not final_image_url:
                                raise Exception(f"第{actual_index}组 - 轮询超时未完成")

                            # 如果成功获取到图片，跳出重试循环
                            if final_image_url:
                                break
                    
                    # 保存图片到 images 表（与原逻辑一致，不放锁里，避免长时间持锁）
                    await self._save_mj_image_to_db(final_image_url, workflow_id, actual_index)

                    # 增量更新 workflow.result_data，使用保存锁串行化写库
                    if self._save_lock is None:
                        self._save_lock = asyncio.Lock()
                    async with self._save_lock:
                        # 验证URL有效性
                        if final_image_url and isinstance(final_image_url, str) and final_image_url.strip():
                            workflow_latest = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
                            latest_result = (workflow_latest.result_data or {}).copy()
                            current_all_images = list(latest_result.get('mj_image_urls') or latest_result.get('mj_image_url') or [])
                            if not isinstance(current_all_images, list):
                                current_all_images = [current_all_images]
                            current_all_images.append(final_image_url)
                            latest_result['mj_image_urls'] = current_all_images
                            ai_workflow.update_workflow_result(db=self.db, workflow_id=workflow_id, result_data=latest_result)
                        else:
                            print(f"⚠️ 第{actual_index}组 - 跳过无效的图片URL: {final_image_url}")

                    print(f"✅ 第{actual_index}组 - MJ生图完成: {final_image_url}")

                except Exception as e:
                    error_msg = f"第{actual_index}组 - MJ 生成失败: {str(e)}"
                    print(f"❌ {error_msg}")
                    
                    # 记录错误到数据库
                    try:
                        from app.crud.ai_workflow import error_history, ai_workflow
                        workflow = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
                        if workflow:
                            error_history.create_error(
                                db=self.db,
                                user_id=workflow.user_id,
                                user_name="",
                                workflow_id=workflow_id,
                                step_id=None,
                                error_type="mj_generation_error",
                                error_message=error_msg,
                                api_provider="MJ"
                            )
                    except Exception as db_error:
                        print(f"⚠️ 记录错误到数据库失败: {str(db_error)}")
                    
                    # 重新抛出异常，让asyncio.gather能够感知到失败
                    raise Exception(error_msg)

            # 并发启动所有 prompts 的任务（并发上限由 KeyPool 控制）
            tasks = []
            for i, current_prompt in enumerate(prompts_to_process):
                actual_index = already_completed + i + 1
                tasks.append(process_one(actual_index, current_prompt))

            if tasks:
                # 使用return_exceptions=True，这样即使某些任务失败，其他任务仍可继续
                results = await asyncio.gather(*tasks, return_exceptions=True)
                
                # 统计成功和失败的任务
                success_count = 0
                failed_count = 0
                for i, result in enumerate(results):
                    if isinstance(result, Exception):
                        failed_count += 1
                        print(f"❌ 任务 {already_completed + i + 1} 失败: {str(result)}")
                    else:
                        success_count += 1
                
                print(f"🔍 并发任务完成统计: 成功 {success_count}/{len(tasks)}, 失败 {failed_count}/{len(tasks)}")
                
                # 如果所有任务都失败了，抛出异常
                if failed_count == len(tasks):
                    raise Exception(f"所有 {len(tasks)} 个MJ生成任务都失败了")
                elif failed_count > 0:
                    print(f"⚠️ 部分任务失败，但有 {success_count} 个任务成功完成")

            # 汇总（兼容原返回）
            workflow_final = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
            all_images = (workflow_final.result_data or {}).get('mj_image_urls') or []
            print(f"🔍 总计: {len(all_images)}张")
            return all_images
            
        except Exception as e:
            print(f"❌ step2_mj_generation 异常: {str(e)}")
            raise Exception(f"MJ文生图失败: {str(e)}")

    async def _poll_mj_status(self, task_id: str, headers: dict, max_attempts: int = 30) -> str:
        """轮询MJ任务状态"""
        status_url = f"https://zeakai-api.api4midjourney.com/mj/task/{task_id}/fetch"
        
        # 添加状态检查函数
        def check_if_cancelled():
            workflow = ai_workflow.get_workflow(db=self.db, workflow_id=self.workflow_id)
            if workflow and workflow.status == TaskStatus.CANCELLED:
                raise Exception("任务已被取消")
        
        # 创建新的HTTP会话
        async with self._get_http_session() as session:
            for attempt in range(max_attempts):
                try:
                    # 在每次轮询前检查任务状态
                    check_if_cancelled()
                    
                    # 使用HTTP会话
                    async with session.get(status_url, headers=headers) as response:
                        if response.status == 200:
                            result = await response.json()
                            
                            if result.get('status') == 'SUCCESS':
                                image_url = result.get('imageUrl')
                                if image_url:
                                    print(f"✅ MJ生图成功: {image_url}")
                                    return image_url.strip()
                                else:
                                    raise Exception(f"任务完成但未获取到图片URL: {result}")
                            elif result.get('status') == 'FAILURE' or result.get('status') == 'FAILED':
                                fail_reason = result.get('failReason') or result.get('error') or '未知错误'
                                raise Exception(f"MJ任务失败: {fail_reason}")
                            elif result.get('status') in ['SUBMITTED', 'IN_PROGRESS', 'PROCESSING']:
                                # 任务还在进行中，等待后重试
                                await asyncio.sleep(10)
                            else:
                                await asyncio.sleep(10)
                        else:
                            await asyncio.sleep(5)
                            
                except Exception as e:
                    if "任务已被取消" in str(e):
                        raise
                    await asyncio.sleep(5)
        
        raise Exception("MJ任务超时")
    
    async def step3_mj_split(self, image_url: str, workflow_id: int) -> list:
        """步骤3: MJ图1拆4 - 支持增量保存"""
        try:
            from app.services.image.processor import ImageProcessor
            from app.services.tencent.cos_service import TencentCOSService
            from app.crud.image import image_crud
            from app.crud.ai_workflow import ai_workflow
            import uuid
            
            print(f"🔍 step3_mj_split 开始处理图片: {image_url}")
            
            # 获取workflow信息用于记录
            workflow = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
            
            # 检查是否有已存在的分割图片（断点续传）
            result_data = workflow.result_data or {}
            existing_split_images = result_data.get('split_images', [])
            
            print(f"🔍 发现已存在的分割图片: {len(existing_split_images)}张")
            if existing_split_images:
                for i, img in enumerate(existing_split_images):
                    print(f"🔍 已有第{i+1}张分割图片: {img.get('url', 'N/A')}")
            
            # 如果已经有4张分割图片，直接返回
            if len(existing_split_images) >= 4:
                print(f"✅ MJ图片分割已全部完成，返回已有图片")
                return existing_split_images
            
            # 1. 下载MJ生成的图片
            processor = ImageProcessor()
            image_bytes = processor.download_image_from_url(image_url)
            print(f"🔍 图片下载成功，大小: {len(image_bytes)} bytes")
            
            # 2. 分割图片为4张
            split_images = processor.split_mj_image(image_bytes)
            print(f"🔍 图片分割完成，共{len(split_images)}张")
            
            # 3. 上传到腾讯云COS并保存到数据库 - 增量保存
            cos_service = TencentCOSService()
            uploaded_images = existing_split_images.copy()  # 从已有图片开始
            
            # 计算需要处理的分割图片（跳过已完成的）
            start_index = len(existing_split_images)
            
            for i in range(start_index, len(split_images)):
                img_bytes = split_images[i]
                actual_index = i + 1  # 实际的分割图片序号（1-4）
                
                try:
                    print(f"\n🔍 开始处理第{actual_index}张分割图片...")
                    
                    # 上传到COS
                    upload_result = cos_service.upload_image(
                        image_data=img_bytes,
                        file_extension='jpg',
                        folder='mj_split'
                    )
                    
                    if upload_result['success']:
                        # 生成唯一的entity_id
                        entity_id = f"mj_split_{workflow_id}_{actual_index}_{uuid.uuid4().hex[:8]}"
                        
                        # 保存到数据库 - 修复字段名称映射
                        image_data = {
                            'entity_id': entity_id,
                            'group_id': f'workflow_{workflow_id}',
                            'pic_name': f'MJ分割图片 {actual_index}/4',  # 修复：使用pic_name而不是original_name
                            'original_filename': f'mj_split_{actual_index}.jpg',  # 修复：使用original_filename
                            'cos_url': upload_result['url'],  # 修复：使用cos_url而不是file_url
                            'cos_key': upload_result['key'],  # 修复：使用cos_key而不是file_path
                            'file_size': len(img_bytes),
                            'file_type': 'image/jpeg',  # 修复：使用file_type而不是mime_type
                            'width': None,  # 可以通过PIL获取，这里简化
                            'height': None,
                            'custom_content': f'MJ图片分割结果 {actual_index}/4',  # 修复：使用custom_content而不是description
                            'tencent_tags': f'{{"tags": ["mj", "split", "workflow_{workflow_id}"]}}',  # 修复：使用tencent_tags而不是tags
                            'tencent_response': f'{{"source": "mj_split", "workflow_id": {workflow_id}, "split_index": {actual_index}, "original_url": "{image_url}"}}',  # 修复：使用tencent_response而不是metadata
                            'user_id': workflow.user_id,  # 添加user_id字段
                            'upload_status': 'success'  # 添加upload_status字段
                        }
                        
                        db_image = image_crud.create_image(db=self.db, image_data=image_data)
                        
                        # 构建返回的图片信息
                        new_image_info = {
                            'id': db_image.id,
                            'entity_id': entity_id,
                            'url': upload_result['url'],
                            'file_name': upload_result['file_name'],
                            'split_index': actual_index
                        }
                        
                        uploaded_images.append(new_image_info)
                        
                        # 🔥 增量保存：立即更新workflow result_data
                        workflow = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
                        result_data = workflow.result_data or {}
                        result_data['split_images'] = uploaded_images
                        ai_workflow.update_workflow_result(db=self.db, workflow_id=workflow_id, result_data=result_data)
                        
                        print(f"✅ 第{actual_index}张分割图片处理成功: {upload_result['url']}")
                        print(f"💾 第{actual_index}张分割图片已立即保存到数据库，当前总数: {len(uploaded_images)}/4")
                        
                    else:
                        error_msg = f"第 {actual_index} 张图片上传失败: {upload_result.get('error', '未知错误')}"
                        print(f"❌ {error_msg}")
                        
                        # 记录错误但继续处理其他图片
                        from app.crud.ai_workflow import error_history
                        error_history.create_error(
                            db=self.db,
                            user_id=workflow.user_id,
                            user_name="",
                            workflow_id=workflow_id,
                            step_id=None,
                            error_type="mj_split_upload_error",
                            error_message=error_msg,
                            api_provider="COS"
                        )
                        
                except Exception as e:
                    error_msg = f"处理第 {actual_index} 张分割图片失败: {str(e)}"
                    print(f"❌ {error_msg}")
                    
                    # 记录错误但继续处理其他图片
                    from app.crud.ai_workflow import error_history
                    error_history.create_error(
                        db=self.db,
                        user_id=workflow.user_id,
                        user_name="",
                        workflow_id=workflow_id,
                        step_id=None,
                        error_type="mj_split_upload_error",
                        error_message=error_msg,
                        api_provider="COS"
                    )
            
            if not uploaded_images:
                raise Exception("所有图片分割上传都失败了")
            
            # 打印最终结果
            urls = [img['url'] for img in uploaded_images]
            print(f"\n✅ MJ图片分割步骤完成:")
            print(f"🔍 成功处理: {len(uploaded_images)}/4 张分割图片")
            print(f"🔍 图片URLs: {urls}")
            
            return uploaded_images
            
        except Exception as e:
            error_msg = f"MJ图片1拆4失败: {str(e)}"
            print(f"❌ {error_msg}")
        
            # 记录错误到error_histories
            from app.crud.ai_workflow import error_history, ai_workflow
            workflow = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
            error_history.create_error(
                db=self.db,
                user_id=workflow.user_id,
                user_name="",
                workflow_id=workflow_id,
                step_id=None,
                error_type="mj_split_error",
                error_message=error_msg,
                api_provider="MJ"
            )
            return None
    
    async def _save_mj_image_to_db(self, image_url: str, workflow_id: int, mj_index: int):
        """保存MJ生成的图片到images表"""
        try:
            from app.services.tencent.cos_service import TencentCOSService
            from app.crud.image import image_crud
            from app.crud.ai_workflow import ai_workflow
            import uuid
            import json
            
            # 获取workflow信息
            workflow = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
            
            # 下载图片
            # 创建新的HTTP会话
            async with self._get_http_session() as session:
                async with session.get(image_url) as response:
                    if response.status == 200:
                        image_bytes = await response.read()
                    else:
                        print(f"❌ 下载MJ图片失败: {response.status}")
                        return
            
            # 上传到腾讯云COS
            cos_service = TencentCOSService()
            upload_result = cos_service.upload_image(
                image_data=image_bytes,
                file_extension='jpg',
                folder='mj_generated'
            )
            
            if upload_result['success']:
                # 生成唯一的entity_id
                entity_id = f"mj_gen_{workflow_id}_{mj_index}_{uuid.uuid4().hex[:8]}"
                
                # 构建tags和response数据
                tencent_tags_data = {
                    "tags": ["mj", "generated", f"workflow_{workflow_id}"],
                    "source": "midjourney",
                    "workflow_id": workflow_id,
                    "mj_index": mj_index
                }
                
                tencent_response_data = {
                    "source": "mj_generation",
                    "workflow_id": workflow_id,
                    "mj_index": mj_index,
                    "original_mj_url": image_url,
                    "upload_timestamp": str(datetime.now())
                }
                
                # 保存到数据库
                image_data = {
                    'entity_id': entity_id,
                    'group_id': f'workflow_{workflow_id}',
                    'pic_name': f'MJ生成图片 {mj_index}',
                    'original_filename': f'mj_gen_{mj_index}.jpg',
                    'cos_url': upload_result['url'],
                    'cos_key': upload_result['key'],
                    'file_size': len(image_bytes),
                    'file_type': 'image/jpeg',
                    'width': None,
                    'height': None,
                    'custom_content': f'MJ生成图片 {mj_index}',
                    'tencent_tags': json.dumps(tencent_tags_data, ensure_ascii=False),
                    'tencent_response': json.dumps(tencent_response_data, ensure_ascii=False),
                    'user_id': workflow.user_id,
                    'upload_status': 'success'
                }
                
                db_image = image_crud.create_image(db=self.db, image_data=image_data)
                print(f"✅ MJ图片 {mj_index} 已保存到images表，ID: {db_image.id}")
                
            else:
                print(f"❌ MJ图片 {mj_index} 上传到COS失败: {upload_result.get('error', '未知错误')}")
                
        except Exception as e:
            print(f"❌ 保存MJ图片 {mj_index} 到数据库失败: {str(e)}")

    async def step4_flux_variation(self, image_urls: list, workflow_id: int) -> list:
        """步骤4: Flux裂变 - 支持增量保存（DB 级幂等 + 缺失索引生成）"""
        try:
            from app.services.ai.flux_service import flux_service
            from app.services.tencent.cos_service import TencentCOSService
            from app.crud.image import image_crud
            from app.crud.ai_workflow import ai_workflow
            import uuid
            
            if not image_urls:
                raise Exception("没有输入图片进行Flux裂变")
            
            print(f"🔍 step4_flux_variation 开始处理 {len(image_urls)} 张输入图片")
            
            # 获取workflow信息
            workflow = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
            
            # 从input_data中获取参数
            input_data = workflow.input_data or {}
            variation_count = input_data.get('variation_count', 4)
            
            # 从result_data中获取第一步的反推提示词
            result_data = workflow.result_data or {}
            optimized_prompt = result_data.get('optimized_prompt', '')
            
            # === 新增：将 optimized_prompt 解析为独立的提示词列表（与 step2 解析规则保持一致） ===
            prompts_list = []
            try:
                lines = (optimized_prompt or "").split('\n')
                for line in lines:
                    line = line.strip()
                    if not line or '生成的英文提示词内容' in line:
                        continue
                    if '反推英文提示词内容' in line:
                        extracted_prompt = line.split('反推英文提示词内容')[-1].strip()
                        if extracted_prompt.startswith('-'):
                            extracted_prompt = extracted_prompt.lstrip('- ').strip()
                        if extracted_prompt and len(extracted_prompt) > 10:
                            prompts_list.append(extracted_prompt)
                            continue
                    if line.startswith('- ') and len(line) > 10:
                        extracted_prompt = line.lstrip('- ').strip()
                        if not any(keyword in extracted_prompt.lower() for keyword in ['生成', '提示词', '内容', '英文']):
                            prompts_list.append(extracted_prompt)
                            continue
                    if line and line[0].isdigit() and ('. ' in line or ') ' in line):
                        if '. ' in line:
                            extracted_prompt = line.split('. ', 1)[1].strip()
                        else:
                            extracted_prompt = line.split(') ', 1)[1].strip()
                        if extracted_prompt and len(extracted_prompt) > 10:
                            prompts_list.append(extracted_prompt)
                            continue
                    if len(line) > 20 and not any('\u4e00' <= c <= '\u9fff' for c in line):
                        english_keywords = ['a', 'an', 'the', 'and', 'or', 'with', 'in', 'on', 'at', 'of', 'for']
                        if any(k in line.lower().split() for k in english_keywords):
                            prompts_list.append(line)
                if not prompts_list:
                    # 兜底：尽量从长行中取值
                    for line in lines:
                        line = line.strip()
                        if line and len(line) > 10 and '生成的英文提示词内容' not in line:
                            prompts_list.append(line)
                if not prompts_list and optimized_prompt:
                    prompts_list = [optimized_prompt.strip()]
            except Exception:
                prompts_list = [optimized_prompt.strip()] if optimized_prompt else []
            
            # 检查是否有已存在的Flux变体（断点续传 - result_data层）
            existing_variations = result_data.get('flux_variations', [])
            print(f"🔍 发现已存在的Flux变体(内存/结果数据): {len(existing_variations)}张")
            if existing_variations:
                for i, var in enumerate(existing_variations):
                    print(f"🔍 已有变体{i+1}: 源图{var.get('source_image_index', 'N/A')}-变体{var.get('variation_index', 'N/A')} - {var.get('url', 'N/A')}")
            
            # 计算总共需要的变体数量
            total_expected = len(image_urls) * variation_count
            print(f"🔍 总共需要: {total_expected}张变体 ({len(image_urls)}张输入图片 × {variation_count}个变体)")
            print(f"🔍 已完成(result_data): {len(existing_variations)}张变体")
            print(f"🔍 还需要(粗略，未考虑DB): {total_expected - len(existing_variations)}张变体")
            
            # ✅ 新增：从 DB 加载已有记录时，同时建立 MJ 与分割图之间的映射，供提示词选择用
            group_id = f'workflow_{workflow_id}'
            db_images = image_crud.get_images_by_group(db=self.db, group_id=group_id, skip=0, limit=2000)
            existing_db_pairs = set()
            db_pair_to_image = {}
            mj_url_to_index = {}          # 原始 MJ 外链URL -> mj_index（用于找到属于哪一组提示词）
            split_cos_to_original = {}    # 分割图 COS URL -> 原始 MJ 外链URL
            for img in db_images:
                try:
                    tr = json.loads(img.tencent_response) if img.tencent_response else {}
                    source = tr.get("source")
                    if source == "flux_variation":
                        s_idx = tr.get("source_image_index")
                        v_idx = tr.get("variation_index")
                        if s_idx and v_idx:
                            key = (int(s_idx), int(v_idx))
                            existing_db_pairs.add(key)
                            db_pair_to_image[key] = img
                    elif source == "mj_generation":
                        orig = tr.get("original_mj_url")
                        mj_idx = tr.get("mj_index")
                        if orig and mj_idx:
                            mj_url_to_index[str(orig)] = int(mj_idx)
                    elif source == "mj_split":
                        orig = tr.get("original_url")
                        if img.cos_url and orig:
                            split_cos_to_original[str(img.cos_url)] = str(orig)
                except Exception:
                    continue
            print(f"🔍 DB中已存在Flux变体: {len(existing_db_pairs)} 个索引对")
            
            # 如果DB层面已全部完成，直接返回DB中的记录
            if len(existing_db_pairs) >= total_expected:
                print("✅ Flux裂变在DB中已全部存在，返回DB记录（强幂等）")
                all_from_db = []
                for (s_idx, v_idx), img in db_pair_to_image.items():
                    all_from_db.append({
                        'id': img.id,
                        'entity_id': getattr(img, 'entity_id', None),
                        'url': img.cos_url,
                        'file_name': getattr(img, 'original_filename', None),
                        'source_image_index': s_idx,
                        'variation_index': v_idx
                    })
                return all_from_db
            
            all_variations = existing_variations.copy()  # 从已有变体开始
            cos_service = TencentCOSService()
            
            # 对每张输入图片进行Flux裂变
            for i, image_data in enumerate(image_urls):
                source_image_index = i + 1
                
                try:
                    # 获取图片URL
                    if isinstance(image_data, dict):
                        image_url = image_data.get('url')
                        original_name = image_data.get('file_name', f'split_{source_image_index}')
                    else:
                        image_url = image_data
                        original_name = f'split_{source_image_index}'
                    
                    if not image_url:
                        print(f"❌ 第{source_image_index}张输入图片URL为空，跳过")
                        continue
                    
                    # === 新增：根据分割图 -> 原始MJ -> mj_index 选择对应提示词 ===
                    associated_mj_index = None
                    try:
                        mj_original_url = split_cos_to_original.get(str(image_url))
                        if mj_original_url and mj_original_url in mj_url_to_index:
                            associated_mj_index = int(mj_url_to_index[mj_original_url])  # 1..10
                    except Exception:
                        associated_mj_index = None
                    if not associated_mj_index:
                        # 兜底：按顺序推断（每4张分割图一组）
                        associated_mj_index = int(math.ceil(source_image_index / 4.0))
                    # 选择该分割图应使用的提示词（1-indexed）
                    if prompts_list and 1 <= associated_mj_index <= len(prompts_list):
                        prompt_for_this_image = prompts_list[associated_mj_index - 1]
                    else:
                        prompt_for_this_image = optimized_prompt or ""
                    
                    # 计算该源图缺失的变体索引（1..variation_count），依据DB层数据
                    existing_for_this_image_db = {v for (s, v) in existing_db_pairs if s == source_image_index}
                    missing_indices = [idx for idx in range(1, variation_count + 1) if idx not in existing_for_this_image_db]
                    
                    if not missing_indices:
                        print(f"🔍 第{source_image_index}张输入图片的变体在DB中已全部存在，跳过")
                        continue
                    
                    print(f"\n🔍 开始处理第{source_image_index}张输入图片: {original_name}")
                    print(f"🔍 DB已存在变体索引: {sorted(list(existing_for_this_image_db))}，缺失索引: {missing_indices}")
                    
                    # 仅为缺失的索引生成对应数量的变体
                    success, variation_images = await flux_service.generate_variations(
                        image_url=image_url,
                        prompt=prompt_for_this_image,  # 使用与该分割图对应的提示词
                        variation_count=len(missing_indices)
                    )
                    
                    # 检查返回结果是否有效
                    if not success or not variation_images:
                        error_msg = f"第 {source_image_index} 张图片的Flux裂变失败或未生成图片"
                        print(f"❌ {error_msg}")
                        from app.crud.ai_workflow import error_history
                        error_history.create_error(
                            db=self.db,
                            user_id=workflow.user_id,
                            user_name="",
                            workflow_id=workflow_id,
                            step_id=None,
                            error_type="flux_variation_error",
                            error_message=error_msg,
                            api_provider="Flux"
                        )
                        continue
                    
                    print(f"🔍 第{source_image_index}张图片Flux裂变成功，生成{len(variation_images)}个新变体（用于缺失索引）")
                    
                    # 逐个上传 - 变体索引映射到缺失索引
                    for j, img_bytes in enumerate(variation_images):
                        variation_index = int(missing_indices[j])
                        
                        try:
                            print(f"\n🔍 开始处理变体 {source_image_index}-{variation_index}...")
                            
                            # 二次幂等保护：上传前再次检查 DB（应对并发）
                            check_key = (source_image_index, variation_index)
                            if check_key in existing_db_pairs:
                                print(f"ℹ️ 变体 {source_image_index}-{variation_index} 在并发过程中已存在于DB，跳过上传")
                                img = db_pair_to_image.get(check_key)
                                if img:
                                    all_variations.append({
                                        'id': img.id,
                                        'entity_id': getattr(img, 'entity_id', None),
                                        'url': img.cos_url,
                                        'file_name': getattr(img, 'original_filename', None),
                                        'source_image_index': source_image_index,
                                        'variation_index': variation_index
                                    })
                                    # 增量保存
                                    workflow_latest = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
                                    rd = workflow_latest.result_data or {}
                                    rd['flux_variations'] = all_variations
                                    ai_workflow.update_workflow_result(db=self.db, workflow_id=workflow_id, result_data=rd)
                                continue
                            
                            # 上传到COS
                            upload_result = cos_service.upload_image(
                                image_data=img_bytes,
                                file_extension='jpg',
                                folder='flux_variations'
                            )
                            
                            if upload_result['success']:
                                # 生成唯一的entity_id
                                entity_id = f"flux_var_{workflow_id}_{source_image_index}_{variation_index}_{uuid.uuid4().hex[:8]}"
                                
                                # 保存到数据库
                                tencent_tags_data = {
                                    "tags": ["flux", "variation", f"workflow_{workflow_id}"],
                                    "source_image": original_name,
                                    "variation_index": variation_index
                                }
                                
                                tencent_response_data = {
                                    "source": "flux_variation",
                                    "workflow_id": workflow_id,
                                    "source_image_index": source_image_index,
                                    "variation_index": variation_index
                                }
                                
                                image_data_db = {
                                    'entity_id': entity_id,
                                    'group_id': f'workflow_{workflow_id}',
                                    'pic_name': f'Flux变体 {source_image_index}-{variation_index}',
                                    'original_filename': f'flux_var_{source_image_index}_{variation_index}.jpg',
                                    'cos_url': upload_result['url'],
                                    'cos_key': upload_result['key'],
                                    'file_size': len(img_bytes),
                                    'file_type': 'image/jpeg',
                                    'width': None,
                                    'height': None,
                                    'custom_content': f'Flux变体图片 {source_image_index}-{variation_index}，基于 {original_name}',
                                    'tencent_tags': json.dumps(tencent_tags_data, ensure_ascii=False),
                                    'tencent_response': json.dumps(tencent_response_data, ensure_ascii=False),
                                    'user_id': workflow.user_id,
                                    'upload_status': 'success'
                                }
                                
                                db_image = image_crud.create_image(db=self.db, image_data=image_data_db)
                                
                                # 更新DB存在集合，避免后续并发重复
                                key = (source_image_index, variation_index)
                                existing_db_pairs.add(key)
                                db_pair_to_image[key] = db_image
                                
                                # 构建返回的变体信息
                                new_variation_info = {
                                    'id': db_image.id,
                                    'entity_id': entity_id,
                                    'url': upload_result['url'],
                                    'file_name': upload_result['file_name'],
                                    'source_image_index': source_image_index,
                                    'variation_index': variation_index
                                }
                                
                                all_variations.append(new_variation_info)
                                
                                # 🔥 增量保存：立即更新workflow result_data
                                workflow_latest = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
                                rd = workflow_latest.result_data or {}
                                rd['flux_variations'] = all_variations
                                ai_workflow.update_workflow_result(db=self.db, workflow_id=workflow_id, result_data=rd)
                                
                                print(f"✅ 变体 {source_image_index}-{variation_index} 处理成功: {upload_result['url']}")
                                print(f"💾 变体 {source_image_index}-{variation_index} 已立即保存到数据库，当前总数: {len(all_variations)}/{total_expected}")
                            else:
                                error_msg = f"Flux变体 {source_image_index}-{variation_index} 上传失败: {upload_result.get('error', '未知错误')}"
                                print(f"❌ {error_msg}")
                                
                                # 记录错误但继续处理
                                from app.crud.ai_workflow import error_history
                                error_history.create_error(
                                    db=self.db,
                                    user_id=workflow.user_id,
                                    user_name="",
                                    workflow_id=workflow_id,
                                    step_id=None,
                                    error_type="flux_variation_upload_error",
                                    error_message=error_msg,
                                    api_provider="Flux"
                                )
                        except Exception as e:
                            error_msg = f"处理Flux变体 {source_image_index}-{variation_index} 失败: {str(e)}"
                            print(f"❌ {error_msg}")
                            
                            # 记录错误但继续处理
                            from app.crud.ai_workflow import error_history
                            error_history.create_error(
                                db=self.db,
                                user_id=workflow.user_id,
                                user_name="",
                                workflow_id=workflow_id,
                                step_id=None,
                                error_type="flux_variation_upload_error",
                                error_message=error_msg,
                                api_provider="Flux"
                            )
                except Exception as e:
                    error_msg = f"处理第 {source_image_index} 张图片的Flux裂变失败: {str(e)}"
                    print(f"❌ {error_msg}")
                    
                    # 记录错误但继续处理其他图片
                    from app.crud.ai_workflow import error_history
                    error_history.create_error(
                        db=self.db,
                        user_id=workflow.user_id,
                        user_name="",
                        workflow_id=workflow_id,
                        step_id=None,
                        error_type="flux_variation_error",
                        error_message=error_msg,
                        api_provider="Flux"
                    )
            
            # 若一个也没有新增，且结果列表为空，报错
            if not all_variations and not existing_db_pairs:
                raise Exception("所有图片的Flux裂变都失败了")
            
            # 合并DB中的已存在记录到返回列表，确保结果全覆盖（去重按索引对）
            existing_pairs_in_list = set()
            for var in all_variations:
                try:
                    s = int(var.get('source_image_index'))
                    v = int(var.get('variation_index'))
                    existing_pairs_in_list.add((s, v))
                except Exception:
                    continue
            for (s_idx, v_idx), img in db_pair_to_image.items():
                if (s_idx, v_idx) not in existing_pairs_in_list:
                    all_variations.append({
                        'id': img.id,
                        'entity_id': getattr(img, 'entity_id', None),
                        'url': img.cos_url,
                        'file_name': getattr(img, 'original_filename', None),
                        'source_image_index': s_idx,
                        'variation_index': v_idx
                    })
            
            # 打印最终结果
            urls = [var['url'] for var in all_variations]
            print(f"\n✅ Flux裂变步骤完成:")
            print(f"🔍 成功处理: {len(all_variations)}/{total_expected} 个变体")
            print(f"🔍 按输入图片分组:")
            for i in range(1, len(image_urls) + 1):
                variations_for_image = [var for var in all_variations if var.get('source_image_index') == i]
                print(f"🔍   第{i}张输入图片: {len(variations_for_image)}/{variation_count} 个变体")
            
            return all_variations
        
        except Exception as e:
            error_msg = f"Flux裂变失败: {str(e)}"
            print(f"❌ {error_msg}")
            
            # 记录错误到error_histories
            from app.crud.ai_workflow import error_history, ai_workflow
            workflow = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
            error_history.create_error(
                db=self.db,
                user_id=workflow.user_id,
                user_name="",
                workflow_id=workflow_id,
                step_id=None,
                error_type="flux_variation_error",
                error_message=error_msg,
                api_provider="Flux"
            )
            return None

    async def step5_generate_product_info(self, optimized_prompt: str, workflow_id: int, resume_from_existing: bool = False, initial_flux_variations: list = None) -> list:
        """步骤5: 智能体生成标题/产品简介/标签 - 并发处理版本"""
        try:
            from app.crud.ai_workflow import ai_workflow
            
            # 获取工作流信息
            workflow = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
            if not workflow:
                raise Exception(f"工作流 {workflow_id} 不存在")
            
            # 统一初始化 result_data，避免在 initial_flux_variations 分支未定义
            result_data = workflow.result_data or {}
            
            # 优先使用传入的 flux_variations，否则从 result_data 获取
            if initial_flux_variations is not None:
                flux_variations = initial_flux_variations
                print(f"🔧 使用传入的 flux_variations 数据，共 {len(flux_variations)} 张图片")
            else:
                # 从result_data中获取flux_variations
                flux_variations = result_data.get('flux_variations', [])
                print(f"🔍 从数据库获取 flux_variations 数据，共 {len(flux_variations)} 张图片")
            
            if not flux_variations:
                print("⚠️ 没有找到Flux变体图片，跳过产品信息生成")
                return []
            
            # 检查是否需要断点续传（兼容 list/dict）
            existing_raw = result_data.get('product_info')
            existing_product_info = []
            if isinstance(existing_raw, list):
                existing_product_info = existing_raw
            elif isinstance(existing_raw, dict):
                try:
                    existing_product_info = sorted(existing_raw.values(), key=lambda x: x.get('image_index', 0))
                except Exception:
                    existing_product_info = list(existing_raw.values())
            else:
                existing_product_info = []

            if resume_from_existing and existing_product_info:
                print(f"🔍 断点续传：已有产品信息 {len(existing_product_info)} 条")
                start_index = len(existing_product_info)
            else:
                existing_product_info = []
                start_index = 0
            
            print(f"🔍 开始并发生成产品信息，从第 {start_index + 1} 张图片开始，共需处理 {len(flux_variations)} 张")
            
            # 如果已经全部完成，直接返回
            if len(existing_product_info) >= len(flux_variations):
                print(f"✅ 产品信息已全部生成完成，返回已有数据")
                return existing_product_info

            # 准备并发任务
            tasks = []
            image_indices = list(range(start_index, len(flux_variations)))
            
            # 为每张图片创建 3 个并发任务（标题、标签、简介）
            for i in image_indices:
                image_data = flux_variations[i]
                image_url = image_data.get('url')
                
                if not image_url:
                    print(f"⚠️ 第 {i+1} 张图片URL为空，跳过")
                    continue
                
                # 创建三个并发任务
                title_task = asyncio.create_task(
                    self._generate_product_title_task(workflow.user_id, image_url, workflow_id, i),
                    name=f"title_{i}"
                )
                tags_task = asyncio.create_task(
                    self._generate_product_tags_task(workflow.user_id, image_url, workflow_id, i),
                    name=f"tags_{i}"
                )
                desc_task = asyncio.create_task(
                    self._generate_product_description_task(workflow.user_id, image_url, workflow_id, i),
                    name=f"desc_{i}"
                )
                
                tasks.extend([title_task, tags_task, desc_task])
            
            if not tasks:
                print("⚠️ 没有创建任何任务，可能所有图片URL都为空")
                return existing_product_info
            
            print(f"🚀 开始执行 {len(tasks)} 个并发任务（{len(tasks)//3} 张图片 × 3 个任务）")
            
            # 执行所有并发任务
            results = await asyncio.gather(*tasks, return_exceptions=True)
            
            # 整理结果
            product_info_list = existing_product_info.copy()
            task_index = 0
            
            for i in image_indices:
                image_data = flux_variations[i]
                image_url = image_data.get('url')
                
                if not image_url:
                    continue
                
                # 获取这张图片的三个任务结果
                title_result = results[task_index] if task_index < len(results) else None
                tags_result = results[task_index + 1] if task_index + 1 < len(results) else None
                desc_result = results[task_index + 2] if task_index + 2 < len(results) else None
                task_index += 3
                
                # 处理结果
                title_success, product_title = self._extract_task_result(title_result, f"第{i+1}张图片标题")
                tags_success, product_tags = self._extract_task_result(tags_result, f"第{i+1}张图片标签")
                desc_success, product_description = self._extract_task_result(desc_result, f"第{i+1}张图片简介")
                
                # 检查是否至少有一个成功
                if not (title_success or tags_success or desc_success):
                    error_msg = f"第 {i+1} 张图片的所有产品信息生成都失败了"
                    print(f"❌ {error_msg}")
                    
                    # 记录错误但继续处理
                    from app.crud.ai_workflow import error_history
                    error_history.create_error(
                        db=self.db,
                        user_id=workflow.user_id,
                        user_name="",
                        workflow_id=workflow_id,
                        step_id=None,
                        error_type="product_info_generation_error",
                        error_message=error_msg,
                        api_provider="Coze"
                    )
                    continue
                
                # 组装产品信息
                product_info = {
                    'image_index': i + 1,
                    'source_image_url': image_url,
                    'source_image_data': image_data,
                    'title': product_title if title_success else '',
                    'tags': product_tags if tags_success else '',
                    'description': product_description if desc_success else '',
                    'title_success': title_success,
                    'tags_success': tags_success,
                    'description_success': desc_success,
                    'generated_at': datetime.utcnow().isoformat()
                }
                
                product_info_list.append(product_info)
                print(f"✅ 第 {i+1} 张图片产品信息已生成，成功项：标题({title_success}), 标签({tags_success}), 简介({desc_success})")
            
            # 批量保存结果
            if product_info_list != existing_product_info:
                workflow = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
                updated_result_data = workflow.result_data or {}
                updated_result_data['product_info'] = product_info_list
                ai_workflow.update_workflow_result(
                    db=self.db,
                    workflow_id=workflow_id,
                    result_data=updated_result_data
                )
            
            # ===== 最终保存兜底 + 打印保存数量 =====
            try:
                wf_check = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
                rd_check = (wf_check.result_data or {}) if wf_check else {}
                pi_check = rd_check.get("product_info")
                if isinstance(pi_check, list):
                    saved_count = len(pi_check)
                elif isinstance(pi_check, dict):
                    saved_count = sum(1 for v in pi_check.values() if v)
                else:
                    saved_count = 0

                final_count = len(product_info_list)
                print(f"📝 Step5 即将返回，DB中已保存数量={saved_count}，本次生成数量={final_count}")

                if saved_count < final_count:
                    # 兜底写入一次
                    updated_result_data = rd_check.copy()
                    updated_result_data["product_info"] = product_info_list
                    ai_workflow.update_workflow_result(db=self.db, workflow_id=workflow_id, result_data=updated_result_data)

                    # 再读一次核验
                    wf_check2 = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
                    rd_check2 = (wf_check2.result_data or {}) if wf_check2 else {}
                    pi_check2 = rd_check2.get("product_info")
                    if isinstance(pi_check2, list):
                        saved_count2 = len(pi_check2)
                    elif isinstance(pi_check2, dict):
                        saved_count2 = sum(1 for v in pi_check2.values() if v)
                    else:
                        saved_count2 = 0
                    print(f"✅ 兜底保存完成，保存后数量={saved_count2}")
                else:
                    print("ℹ️ 兜底保存不需要，数据库数量已达到或超过本次生成数量")
            except Exception as e:
                print(f"❌ 兜底保存失败: {e}")

            return product_info_list
            
        except Exception as e:
            print(f"❌ Step5 执行异常: {e}")
            return []

    async def _generate_product_title_task(self, user_id: int, image_url: str, workflow_id: int, image_index: int) -> Tuple[bool, str]:
        """生成产品标题任务"""
        try:
            return await coze_service.process_product_title_async(
                user_id=user_id,
                image_url=image_url,
                db=self.db,
                workflow_id=workflow_id
            )
        except Exception as e:
            print(f"❌ 第{image_index+1}张图片标题生成异常: {str(e)}")
            return False, f"标题生成异常: {str(e)}"

    async def _generate_product_tags_task(self, user_id: int, image_url: str, workflow_id: int, image_index: int) -> Tuple[bool, str]:
        """生成产品标签任务"""
        try:
            return await coze_service.process_product_tags_async(
                user_id=user_id,
                image_url=image_url,
                db=self.db,
                workflow_id=workflow_id
            )
        except Exception as e:
            print(f"❌ 第{image_index+1}张图片标签生成异常: {str(e)}")
            return False, f"标签生成异常: {str(e)}"

    async def _generate_product_description_task(self, user_id: int, image_url: str, workflow_id: int, image_index: int) -> Tuple[bool, str]:
        """生成产品简介任务"""
        try:
            return await coze_service.process_product_description_async(
                user_id=user_id,
                image_url=image_url,
                db=self.db,
                workflow_id=workflow_id
            )
        except Exception as e:
            print(f"❌ 第{image_index+1}张图片简介生成异常: {str(e)}")
            return False, f"简介生成异常: {str(e)}"

    def _extract_task_result(self, result, task_name: str) -> Tuple[bool, str]:
        """提取任务结果"""
        if isinstance(result, Exception):
            print(f"❌ {task_name}任务异常: {str(result)}")
            return False, f"任务异常: {str(result)}"
        elif isinstance(result, tuple) and len(result) == 2:
            success, content = result
            if success:
                print(f"✅ {task_name}成功: {content[:50]}..." if len(content) > 50 else f"✅ {task_name}成功: {content}")
            else:
                print(f"❌ {task_name}失败: {content}")
            return success, content
        else:
            print(f"❌ {task_name}返回格式错误: {result}")
            return False, f"返回格式错误: {result}"

    async def _generate_product_title_task(self, user_id: int, image_url: str, workflow_id: int, image_index: int) -> Tuple[bool, str]:
        """生成产品标题任务"""
        try:
            return await coze_service.process_product_title_async(
                user_id=user_id,
                image_url=image_url,
                db=self.db,
                workflow_id=workflow_id
            )
        except Exception as e:
            print(f"❌ 第{image_index+1}张图片标题生成异常: {str(e)}")
            return False, f"标题生成异常: {str(e)}"

    async def _generate_product_tags_task(self, user_id: int, image_url: str, workflow_id: int, image_index: int) -> Tuple[bool, str]:
        """生成产品标签任务"""
        try:
            return await coze_service.process_product_tags_async(
                user_id=user_id,
                image_url=image_url,
                db=self.db,
                workflow_id=workflow_id
            )
        except Exception as e:
            print(f"❌ 第{image_index+1}张图片标签生成异常: {str(e)}")
            return False, f"标签生成异常: {str(e)}"

    async def _generate_product_description_task(self, user_id: int, image_url: str, workflow_id: int, image_index: int) -> Tuple[bool, str]:
        """生成产品简介任务"""
        try:
            return await coze_service.process_product_description_async(
                user_id=user_id,
                image_url=image_url,
                db=self.db,
                workflow_id=workflow_id
            )
        except Exception as e:
            print(f"❌ 第{image_index+1}张图片简介生成异常: {str(e)}")
            return False, f"简介生成异常: {str(e)}"

    def _extract_task_result(self, result, task_name: str) -> Tuple[bool, str]:
        """提取任务结果"""
        if isinstance(result, Exception):
            print(f"❌ {task_name}任务异常: {str(result)}")
            return False, f"任务异常: {str(result)}"
        elif isinstance(result, tuple) and len(result) == 2:
            success, content = result
            if success:
                print(f"✅ {task_name}成功: {content[:50]}..." if len(content) > 50 else f"✅ {task_name}成功: {content}")
            else:
                print(f"❌ {task_name}失败: {content}")
            return success, content
        else:
            print(f"❌ {task_name}返回格式错误: {result}")
            return False, f"返回格式错误: {result}"


    async def step6_generate_detail_page(self, product_info: list, images: list, workflow_id: int, resume_from_existing: bool = False) -> list:
        """步骤6: GPT4o生成详情页（并发 + Key池 + 串行写库）"""
        try:
            # 获取工作流信息
            workflow = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
            if not workflow:
                raise Exception(f"工作流 {workflow_id} 不存在")

            # 从result_data中获取flux_variations和product_info
            result_data = workflow.result_data or {}

            flux_variations = result_data.get('flux_variations', [])
            product_info_list = result_data.get('product_info', [])

            if not flux_variations:
                print("⚠️ 没有找到Flux变体图片，跳过详情页生成")
                return []

            if not product_info_list:
                print("⚠️ 没有找到产品信息，跳过详情页生成")
                return []

            # 硬上限：不管任何情况，详情页数量不能超过预期（按一一对应关系）
            expected_total = min(len(flux_variations), len(product_info_list))
            if expected_total <= 0:
                print("⚠️ 预期详情页数量为0，跳过详情页生成")
                return []

            # 检查是否需要断点续传
            existing_detail_pages = result_data.get('detail_pages', [])
            if resume_from_existing and existing_detail_pages:
                print(f"🔍 断点续传：已有详情页 {len(existing_detail_pages)} 页")
                # 如果已达上限，直接返回前 expected_total 条，避免超过上限
                if len(existing_detail_pages) >= expected_total:
                    print(f"✅ 已达到或超过详情页上限 {expected_total}，不再生成新详情页")
                    return existing_detail_pages[:expected_total]
                start_index = len(existing_detail_pages)
            else:
                existing_detail_pages = []
                start_index = 0

            # 仅处理“缺口”数量，绝不超过上限
            remaining_needed = max(0, expected_total - start_index)
            print(f"🔍 开始生成详情页，从第 {start_index + 1} 张图片开始，计划处理 {remaining_needed}/{expected_total} 张（硬上限）")

            # 移除跨事件循环持有的锁与Key池，改为在每次任务中获取
            detail_pages_list = existing_detail_pages.copy()
            success_count = 0
            error_count = 0

            async def process_one(i: int):
                nonlocal success_count, error_count, detail_pages_list
                # 导入ai_workflow模块
                from app.crud.ai_workflow import ai_workflow
                
                # 额外保护：在任务函数内部也校验硬上限
                if i >= expected_total:
                    return
                try:
                    flux_image = flux_variations[i]
                    corresponding_product_info = product_info_list[i] if i < len(product_info_list) else None
                    if not corresponding_product_info:
                        print(f"⚠️ 第 {i+1} 张图片没有对应的产品信息，跳过")
                        return

                    print(f"\n🔍 开始为第 {i+1} 张图片生成详情页...")
                    print(f"🔍 图片URL: {flux_image.get('url', 'N/A')}")
                    print(f"🔍 产品标题: {corresponding_product_info.get('title', 'N/A')}")

                    # 从当前事件循环获取 KeyPool（per-loop 实例）
                    key_pool = get_zeakai_key_pool()
                    async with key_pool.acquire() as auth_token:
                        detail_page_content = await self._generate_detail_page_with_gpt4o(
                            image_url=flux_image.get('url', ''),
                            product_info=corresponding_product_info,
                            workflow_id=workflow_id,
                            max_retries=3,
                            auth_token=auth_token
                        )

                    if detail_page_content:
                        # 将详情页中的图片链接下载后上传COS，并用COS链接替换；同时写入 images 表并返回写入结果
                        try:
                            processed_content, saved_images = self._rewrite_detail_content_images_to_cos(
                                content=detail_page_content,
                                workflow_id=workflow_id
                            )
                        except Exception as e:
                            print(f"⚠️ 详情页图片替换为COS链接失败，继续使用原内容: {str(e)}")
                            processed_content = detail_page_content
                            saved_images = []

                        detail_page_data = {
                            "image_url": flux_image.get('url', ''),
                            "image_index": i,
                            "product_title": corresponding_product_info.get('title', ''),
                            "product_tags": corresponding_product_info.get('tags', []),
                            "product_description": corresponding_product_info.get('description', ''),
                            "detail_content": processed_content,
                            "images": saved_images,
                            "generated_at": datetime.utcnow().isoformat()
                        }

                        # 在当前事件循环里首次创建保存锁
                        if self._save_lock is None:
                            self._save_lock = asyncio.Lock()

                        # 串行保护写库，避免并发覆盖（与 step4 相同的增量写回模式）
                        async with self._save_lock:
                            # 在锁内再次校验是否已达上限（防止竞态导致超额）
                            workflow_latest = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
                            latest_result = (workflow_latest.result_data or {}).copy()
                            latest_pages = list(latest_result.get('detail_pages', []))
                            if len(latest_pages) >= expected_total:
                                print(f"⚠️ 已在保存时达到上限 {expected_total}，跳过当前详情页写入")
                                return

                            latest_pages.append(detail_page_data)
                            latest_result['detail_pages'] = latest_pages

                            ai_workflow.update_workflow_result(
                                db=self.db,
                                workflow_id=workflow_id,
                                result_data=latest_result
                            )

                            detail_pages_list.append(detail_page_data)
                            success_count += 1

                        print(f"✅ 第 {i+1} 张图片详情页生成成功并已保存")
                    else:
                        error_count += 1
                        print(f"❌ 第 {i+1} 张图片详情页生成失败")
                except Exception as e:
                    error_count += 1
                    print(f"❌ 第 {i+1} 张图片详情页生成异常: {str(e)}")

            # 并发调度：仅调度“缺口”数量的任务；并发限制由 key 池控制（per-key 和全局）
            task_indices = list(range(start_index, start_index + remaining_needed))
            tasks = [process_one(i) for i in task_indices]
            if tasks:
                await asyncio.gather(*tasks)

            print(f"\n📊 详情页生成完成统计:")
            print(f"✅ 成功: {success_count} 张")
            print(f"❌ 失败: {error_count} 张")
            print(f"📝 总计: {len(detail_pages_list)} 页详情页（上限 {expected_total}）")

            return detail_pages_list

        except Exception as e:
            error_msg = f"详情页生成过程异常: {str(e)}"
            print(f"❌ {error_msg}")
            raise Exception(error_msg)

    def _rewrite_detail_content_images_to_cos(self, content: str, workflow_id: int) -> Tuple[str, list]:
        """
        将详情页内容中的图片链接统一下载并上传至腾讯云COS，然后替换为COS链接。
        支持两类常见格式：
        - HTML: <img src="...">
        - Markdown: ![alt](url)
        同时把成功上传并写入 images 表的图片信息返回给上层，用于增量写回。
        """
        if not content or not isinstance(content, str):
            return content, []

        # 收集图片URL（HTML <img> 与 Markdown 图片）
        img_src_pattern = re.compile(r'<img[^>]+src=["\'](.*?)["\']', re.IGNORECASE)
        md_img_pattern = re.compile(r'!\[[^\]]*\]\(([^)]+)\)')

        urls = []
        urls.extend(img_src_pattern.findall(content) or [])
        urls.extend(md_img_pattern.findall(content) or [])

        # 去重但保持顺序
        seen = set()
        ordered_urls = []
        for u in urls:
            if u not in seen:
                seen.add(u)
                ordered_urls.append(u)

        if not ordered_urls:
            return content, []

        cos = TencentCOSService()
        processor = ImageProcessor()
        workflow = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)

        def is_cos_url(u: str) -> bool:
            return "myqcloud.com" in u

        def is_data_url(u: str) -> bool:
            return u.strip().lower().startswith("data:")

        def normalize_ext(ext: str) -> str:
            ext = (ext or "").lstrip(".").lower()
            if ext in ("jpeg", "jpg"):
                return "jpeg"
            if ext in ("png", "webp", "gif", "bmp"):
                return ext
            # 默认使用 jpeg
            return "jpeg"

        def mime_from_ext(ext: str) -> str:
            mapping = {
                "jpeg": "image/jpeg",
                "jpg": "image/jpeg",
                "png": "image/png",
                "webp": "image/webp",
                "gif": "image/gif",
                "bmp": "image/bmp",
            }
            return mapping.get(ext, "image/jpeg")

        saved_images = []
        processed_content = content
        idx = 0

        for u in ordered_urls:
            try:
                if not u or is_cos_url(u) or is_data_url(u):
                    continue

                # 下载图片
                img_bytes = processor.download_image_from_url(u)
                if not img_bytes:
                    continue

                # 推断扩展名
                try:
                    ext = processor.get_file_extension_from_url(u)
                except Exception:
                    ext = None
                ext_norm = normalize_ext(ext)

                # 上传到COS（以 workflow 维度归档）
                cos_result = cos.upload_image(
                    image_data=img_bytes,
                    file_extension=ext_norm,
                    folder=f"detail_pages/{workflow_id}"
                )
                cos_url = cos_result.get("url") or ""
                cos_key = cos_result.get("key") or ""
                file_name = cos_result.get("file_name") or f"detail_{uuid.uuid4().hex}.{ext_norm}"
                if not cos_url:
                    continue

                # 写入 images 表
                try:
                    entity_id = f"detail_img_{workflow_id}_{uuid.uuid4().hex[:8]}"
                    tencent_tags_data = {
                        "tags": ["detail", f"workflow_{workflow_id}"],
                        "source": "detail_page",
                        "workflow_id": workflow_id
                    }
                    tencent_response_data = {
                        "source": "detail_page",
                        "workflow_id": workflow_id,
                        "original_url": u,
                        "upload_timestamp": str(datetime.now())
                    }
                    image_data_db = {
                        "entity_id": entity_id,
                        "group_id": f"workflow_{workflow_id}",
                        "pic_name": f"详情页图片 {idx+1}",
                        "original_filename": file_name,
                        "cos_url": cos_url,
                        "cos_key": cos_key,
                        "file_size": len(img_bytes),
                        "file_type": mime_from_ext(ext_norm),
                        "width": None,
                        "height": None,
                        "custom_content": f"详情页内容内联图片（源：{u}）",
                        "tencent_tags": json.dumps(tencent_tags_data, ensure_ascii=False),
                        "tencent_response": json.dumps(tencent_response_data, ensure_ascii=False),
                        "user_id": (workflow.user_id if workflow else 1),
                        "upload_status": "success",
                    }
                    db_image = image_crud.create_image(db=self.db, image_data=image_data_db)

                    saved_images.append({
                        "id": getattr(db_image, "id", None),
                        "entity_id": entity_id,
                        "url": cos_url,
                        "file_name": file_name,
                        "original_url": u
                    })
                    idx += 1
                except Exception as e_db:
                    print(f"⚠️ 详情页图片写入 images 表失败（保留继续流程）: {str(e_db)}")

                # 将内容中的原URL替换为COS URL（替换所有出现的该 URL）
                processed_content = re.sub(re.escape(u), cos_url, processed_content)

            except Exception as e:
                # 单个URL失败不影响整体流程，保留原URL
                print(f"⚠️ 图片链接处理失败（保留原链接）: {u} -> {str(e)}")
                continue

        return processed_content, saved_images

    async def execute_batch_pipeline(self, workflow_id: int, image_urls: list, user_prompt: str, variation_count: int = 4):
        """执行批量图片的AI处理管道 - 每张图片独立运行完整流程"""
        try:
            # 添加状态检查函数
            def check_if_cancelled():
                workflow = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
                if workflow and workflow.status == TaskStatus.CANCELLED:
                    raise Exception("任务已被取消")
            
            # 更新状态为处理中
            ai_workflow.update_workflow_status(
                db=self.db, 
                workflow_id=workflow_id, 
                status=TaskStatus.RUNNING,
                current_step="开始批量处理",
                progress=0
            )
            
            # 初始化批量处理的result_data
            result_data = {
                "batch_info": {
                    "total_images": len(image_urls),
                    "variation_count": variation_count,
                    "processed_images": 0
                },
                "processed_images": [],  # 添加这个数组来存储每张图片的完整结果
                "batch_results": [],     # 保留用于兼容性
                "optimized_prompt": None,  # 将存储第一张图片的优化提示词
                "mj_image_urls": [],
                "split_images": [],
                "flux_variations": [],
                "product_info": [],
                "detail_pages": []
            }
            
            # 为每张图片执行完整的独立处理流程
            total_images = len(image_urls)
            for i, image_url in enumerate(image_urls):
                check_if_cancelled()
                
                current_progress = (i * 90 // total_images)  # 0-90%的进度
                
                print(f"🔍 开始处理第{i+1}/{total_images}张图片: {image_url}")
                
                try:
                    # ===== 第一组：反推提示词 + MJ生成 =====
                    ai_workflow.update_workflow_status(
                        db=self.db, 
                        workflow_id=workflow_id,
                        status=TaskStatus.RUNNING,
                        current_step=f"第{i+1}张图片 - 反推提示词",
                        progress=current_progress
                    )
                    
                    print(f"🔍 第{i+1}张图片 - 步骤1: 优化提示词")
                    optimized_prompt = await self.step1_optimize_prompt(image_url, user_prompt, workflow_id)
                    
                    # 如果是第一张图片，保存优化提示词到主结果中
                    if i == 0:
                        # ✅ 重新获取最新的 result_data，避免覆盖
                        workflow = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
                        result_data = workflow.result_data or {
                            "batch_info": {
                                "total_images": len(image_urls),
                                "variation_count": variation_count,
                                "processed_images": 0
                            },
                            "processed_images": [],
                            "batch_results": [],
                            "optimized_prompt": None,
                            "mj_image_urls": [],
                            "split_images": [],
                            "flux_variations": [],
                            "product_info": [],
                            "detail_pages": []
                        }
                        result_data["optimized_prompt"] = optimized_prompt
                        ai_workflow.update_workflow_result(db=self.db, workflow_id=workflow_id, result_data=result_data)
                    
                    ai_workflow.update_workflow_status(
                        db=self.db, 
                        workflow_id=workflow_id,
                        status=TaskStatus.RUNNING,
                        current_step=f"第{i+1}张图片 - MJ生成",
                        progress=current_progress + 10
                    )
                    
                    print(f"🔍 第{i+1}张图片 - 步骤2: MJ文生图")
                    mj_images = await self.step2_mj_generation(optimized_prompt, workflow_id, image_url, resume_from_existing=False)
                    
                    # ❌ 移除这里的重复保存，让 step2_mj_generation 内部的增量保存生效
                    # result_data["mj_image_urls"].extend(mj_images)
                    # ai_workflow.update_workflow_result(db=self.db, workflow_id=workflow_id, result_data=result_data)
                    
                    # ✅ 重新获取最新数据，确保包含 step2 的增量保存结果
                    workflow = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
                    result_data = workflow.result_data or {}
                    print(f"💾 第{i+1}张图片的反推提示词+MJ生成已完成，当前MJ图片总数: {len(result_data.get('mj_image_urls', []))}")
                    
                    # ===== 第二组：图片分割 + Flux变体生成 =====
                    ai_workflow.update_workflow_status(
                        db=self.db, 
                        workflow_id=workflow_id,
                        status=TaskStatus.RUNNING,
                        current_step=f"第{i+1}张图片 - 图片分割",
                        progress=current_progress + 20
                    )
                    
                    print(f"🔍 第{i+1}张图片 - 步骤3: 分割图片")
                    all_split_images = []
                    # 过滤掉None值和无效的URL
                    valid_mj_images = [url for url in mj_images if url and isinstance(url, str) and url.strip()]
                    print(f"🔍 有效的MJ图片URL数量: {len(valid_mj_images)}/{len(mj_images)}")
                    
                    for j, mj_image_url in enumerate(valid_mj_images):
                        try:
                            split_images = await self.step3_mj_split(mj_image_url, workflow_id)
                            if split_images:
                                all_split_images.extend(split_images)
                        except Exception as e:
                            print(f"❌ 第{i+1}张图片的第{j+1}张MJ图片拆分失败: {str(e)}")
                            continue
                    
                    # ❌ 移除这里的重复保存，让 step3_mj_split 内部的增量保存生效
                    # result_data["split_images"].extend(all_split_images)
                    
                    ai_workflow.update_workflow_status(
                        db=self.db, 
                        workflow_id=workflow_id,
                        status=TaskStatus.RUNNING,
                        current_step=f"第{i+1}张图片 - Flux变体生成",
                        progress=current_progress + 30
                    )
                    
                    print(f"🔍 第{i+1}张图片 - 步骤4: Flux变体生成")
                    flux_variations = await self.step4_flux_variation(all_split_images, workflow_id)
                    
                    # ❌ 移除这里的重复保存，让 step4_flux_variation 内部的增量保存生效
                    # result_data["flux_variations"].extend(flux_variations)
                    
                    # ✅ 重新获取最新数据，确保包含 step3 和 step4 的增量保存结果
                    workflow = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
                    result_data = workflow.result_data or {}
                    print(f"💾 第{i+1}张图片的图片分割+Flux变体已完成，当前分割图片总数: {len(result_data.get('split_images', []))}，Flux变体总数: {len(result_data.get('flux_variations', []))}")
                    
                    # ===== 第三组：商品信息生成 + 详情页生成 =====
                    ai_workflow.update_workflow_status(
                        db=self.db, 
                        workflow_id=workflow_id,
                        status=TaskStatus.RUNNING,
                        current_step=f"第{i+1}张图片 - 商品信息生成",
                        progress=current_progress + 40
                    )
                    
                    print(f"🔍 第{i+1}张图片 - 步骤5: 生成商品信息")
                    product_info = await self.step5_generate_product_info(optimized_prompt, workflow_id, resume_from_existing=False)
                    
                    # ❌ 移除这里的重复保存，让 step5_generate_product_info 内部的增量保存生效
                    # result_data["product_info"].extend(product_info)
                    
                    ai_workflow.update_workflow_status(
                        db=self.db, 
                        workflow_id=workflow_id,
                        status=TaskStatus.RUNNING,
                        current_step=f"第{i+1}张图片 - 详情页生成",
                        progress=current_progress + 50
                    )
                    
                    print(f"🔍 第{i+1}张图片 - 步骤6: 生成详情页")
                    detail_pages = await self.step6_generate_detail_page(product_info, flux_variations, workflow_id, resume_from_existing=False)
                    
                    # ❌ 移除这里的重复保存，让 step6_generate_detail_page 内部的增量保存生效
                    # result_data["detail_pages"].extend(detail_pages)
                    
                    # ✅ 重新获取最新数据，确保包含所有步骤的增量保存结果
                    workflow = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
                    result_data = workflow.result_data or {}
                    
                    # 保存每张图片的完整处理结果到 processed_images
                    image_result = {
                        "image_url": image_url,
                        "optimized_prompt": optimized_prompt,  # 完整的提示词文本
                        "mj_image_urls": mj_images,           # 当前图片的MJ结果
                        "split_images": all_split_images,      # 当前图片的分割结果
                        "flux_variations": flux_variations,    # 当前图片的Flux变体
                        "product_info": product_info,          # 当前图片的商品信息
                        "detail_pages": detail_pages,          # 当前图片的详情页
                        "status": "success"
                    }
                    
                    # ✅ 确保 processed_images 数组存在
                    if "processed_images" not in result_data:
                        result_data["processed_images"] = []
                    result_data["processed_images"].append(image_result)
                    
                    # 记录单张图片的处理结果（保留用于兼容性）
                    batch_result = {
                        "image_url": image_url,
                        "optimized_prompt": optimized_prompt,  # 每张图片独立的优化提示词
                        "mj_count": len(mj_images),
                        "split_count": len(all_split_images),
                        "flux_count": len(flux_variations),
                        "product_info_count": len(product_info),
                        "detail_pages_count": len(detail_pages),
                        "status": "success"
                    }
                    
                    # ✅ 确保 batch_results 数组存在
                    if "batch_results" not in result_data:
                        result_data["batch_results"] = []
                    result_data["batch_results"].append(batch_result)
                    
                    # ✅ 确保 batch_info 存在
                    if "batch_info" not in result_data:
                        result_data["batch_info"] = {"processed_images": 0}
                    result_data["batch_info"]["processed_images"] = i + 1
                    
                    # 最终保存该图片的完整结果（只更新 processed_images 和 batch_results，不覆盖全局累积数据）
                    ai_workflow.update_workflow_result(db=self.db, workflow_id=workflow_id, result_data=result_data)
                    print(f"✅ 第{i+1}张图片处理完成并保存")
                    
                except Exception as e:
                    error_msg = f"第{i+1}张图片处理失败: {str(e)}"
                    print(f"❌ {error_msg}")
                    
                    # ✅ 重新获取最新数据，确保不覆盖已有结果
                    workflow = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
                    result_data = workflow.result_data or {}
                    
                    # 处理失败的图片
                    error_result = {
                        "image_url": image_url,
                        "error": error_msg,
                        "status": "failed"
                    }
                    
                    # ✅ 确保 processed_images 数组存在
                    if "processed_images" not in result_data:
                        result_data["processed_images"] = []
                    result_data["processed_images"].append(error_result)
                    
                    # 记录失败的图片（保留用于兼容性）
                    batch_result = {
                        "image_url": image_url,
                        "error": error_msg,
                        "status": "failed"
                    }
                    
                    # ✅ 确保 batch_results 数组存在
                    if "batch_results" not in result_data:
                        result_data["batch_results"] = []
                    result_data["batch_results"].append(batch_result)
                    
                    # ✅ 确保 batch_info 存在
                    if "batch_info" not in result_data:
                        result_data["batch_info"] = {"processed_images": 0}
                    result_data["batch_info"]["processed_images"] = i + 1
                    
                    # 保存错误结果（只更新 processed_images 和 batch_results，不覆盖全局累积数据）
                    ai_workflow.update_workflow_result(db=self.db, workflow_id=workflow_id, result_data=result_data)
            
            # 完成批量处理
            ai_workflow.update_workflow_status(
                db=self.db, 
                workflow_id=workflow_id,
                status=TaskStatus.COMPLETED,
                current_step="批量处理完成",
                progress=100
            )
            
            # ✅ 重新获取最新数据以获取准确的统计结果
            workflow = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
            result_data = workflow.result_data or {}
            
            # 统计处理结果
            successful_count = len([r for r in result_data.get("batch_results", []) if r["status"] == "success"])
            failed_count = len([r for r in result_data.get("batch_results", []) if r["status"] == "failed"])
            
            print(f"✅ 批量工作流 {workflow_id} 执行完成")
            print(f"📊 处理统计: 成功 {successful_count} 张，失败 {failed_count} 张，总计 {total_images} 张")
            
            # 返回最新的完整结果数据
            return result_data
            
        except Exception as e:
            error_msg = f"批量工作流执行失败: {str(e)}"
            print(f"❌ {error_msg}")
            
            ai_workflow.update_workflow_status(
                db=self.db, 
                workflow_id=workflow_id,
                status=TaskStatus.FAILED,
                error_message=error_msg
            )
            raise Exception(error_msg)

    async def _generate_detail_page_with_gpt4o(self, image_url: str, product_info: dict, workflow_id: int, max_retries: int = 3, auth_token: str = None) -> str:
        """使用GPT4o生成详情页内容 - 使用正确的API端点和Authorization请求头；支持外部传入auth_token（来自Key池）"""
        for attempt in range(max_retries):
            try:
                import os

                # 优先使用传入的 token，其次使用环境变量回退
                token = auth_token or os.getenv("OPENAI_API_KEY", "6a375dc8-cddb-4276-9cae-e67247fee125")

                # 构建请求头
                headers = {
                    'Authorization': f'Bearer {token}',
                    'Content-Type': 'application/json'
                }

                # 构建详情页生成的提示词（固定提示词）
                fixed_prompt = "A vertical 2:3 ratio product introduction poster with a clean and structured minimalist design. The background is pure white or soft neutral tones, with subtle dividing lines and soft shadows to separate sections. The layout is divided into three main sections from top to bottom: Top Section (Lifestyle Scene): A wide horizontal lifestyle image showing the product naturally integrated into a real usage environment. The scene should highlight the product’s style and function, with a short English caption in modern sans-serif font. Middle Section (Product Showcase): Three product images placed side by side in a neat layout: two full product shots from different angles and one close-up detail image focusing on functionality. Each image has a small English label beneath it, indicating the corresponding feature or angle. Bottom Section (Product Description): A short block of English text introducing the product’s design style, features, and key highlights. Typography hierarchy: large bold title at the top, medium-sized subtitle below it, and smaller body text for the description. The text should be concise, fluent, and well-balanced with the images, avoiding overcrowding. Overall style: modern minimalist, visually clear and elegant, suitable for an e-commerce product page. The final poster should balance images and text for readability, with structured composition and professional visual hierarchy."

                # 构建请求数据（按照您提供的格式）
                payload = {
                    "model": "gpt-image-1",
                    "messages": [
                        {
                            "role": "user",
                            "content": [
                                {
                                    "type": "text",
                                    "text": fixed_prompt
                                },
                                {
                                    "type": "image_url",
                                    "image_url": {
                                        "url": image_url
                                    }
                                }
                            ]
                        }
                    ],
                    "stream": False
                }

                print(f"🔍 调用GPT4o生成详情页 (尝试 {attempt + 1}/{max_retries})...")

                # 使用HTTP会话上下文管理器发送请求
                timeout = aiohttp.ClientTimeout(total=60 * 10)  # 10分钟超时
                async with self._get_http_session() as session:
                    async with session.post(
                        "https://zeakai-api.api4midjourney.com/v1/chat/completions",  # 使用正确的API端点
                        headers=headers,
                        json=payload,
                        timeout=timeout
                    ) as response:
                        if response.status == 200:
                            result = await response.json()
                            if result.get('choices') and result['choices'][0].get('message', {}).get('content'):
                                detail_content = result['choices'][0]['message']['content'].strip()
                                print(f"✅ GPT4o详情页生成成功，内容长度: {len(detail_content)} 字符")
                                return detail_content
                            else:
                                print("❌ GPT4o返回空内容")
                                if attempt < max_retries - 1:
                                    await asyncio.sleep(2 ** attempt)
                                    continue
                                return None
                        else:
                            error_text = await response.text()
                            raise Exception(f"HTTP {response.status}: {error_text}")

            except Exception as e:
                error_msg = f"GPT4o API调用失败 (尝试 {attempt + 1}/{max_retries}): {str(e)}"
                print(f"❌ {error_msg}")

                if attempt < max_retries - 1:
                    await asyncio.sleep(2 ** attempt)  # 指数退避
                    continue
                else:
                    # 记录错误到数据库
                    try:
                        from app.crud.ai_workflow import error_history
                        workflow = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
                        error_history.create_error(
                            db=self.db,
                            user_id=workflow.user_id if workflow else 1,
                            workflow_id=workflow_id,
                            error_type="GPT4o_API_Error",
                            error_message=error_msg,
                            api_provider="ZeakaiAPI"
                        )
                    except:
                        pass
                    return None

        return None

    async def resume_workflow(self, workflow_id: int) -> dict:
        """断点续传：检查工作流状态并从失败点继续执行"""
        try:
            from app.crud.ai_workflow import ai_workflow
            
            # 获取工作流信息
            workflow = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
            if not workflow:
                raise Exception(f"未找到工作流 ID: {workflow_id}")
            
            print(f"🔍 开始断点续传，工作流ID: {workflow_id}")
            print(f"🔍 当前状态: {workflow.status}")
            print(f"🔍 当前步骤: {workflow.current_step}")
            print(f"🔍 进度: {workflow.progress}%")
            
            # 解析已有的结果数据
            result_data = workflow.result_data or {}
            print(f"🔍 已有结果数据: {list(result_data.keys())}")
            
            # 检查各步骤完成情况 - 传入workflow_id
            steps_status = self._check_steps_completion(result_data, workflow_id)
            
            # 从第一个未完成的步骤开始执行
            return await self._execute_from_step(workflow_id, steps_status, result_data)
            
        except Exception as e:
            error_msg = f"断点续传失败: {str(e)}"
            print(f"❌ {error_msg}")
            raise Exception(error_msg)
    
    def _check_steps_completion(self, result_data: dict, workflow_id: int = None) -> dict:
        """检查各步骤的完成情况 - 根据预期数量判断"""
        from app.crud.ai_workflow import ai_workflow
        
        # 获取工作流信息以计算预期数量
        if workflow_id:
            workflow = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
            input_data = workflow.input_data or {}
            variation_count = input_data.get('variation_count', 4)
            
            # 从input_data获取image_path数组长度，或默认为1（单张图片）
            image_path = input_data.get('image_path', [])
            if isinstance(image_path, list) and len(image_path) > 0:
                image_count = len(image_path)
            else:
                image_count = 1  # 默认值，单张图片处理
        else:
            variation_count = 4  # 默认值
            image_count = 1  # 默认值
        
        # 计算预期数量 - 基于image_count
        base_optimized_prompts = 10 * image_count
        base_mj_images = base_optimized_prompts
        base_split_images = base_mj_images * 4
        
        expected_counts = {
            "optimized_prompts": base_optimized_prompts,
            "mj_images": base_mj_images,
            "split_images": base_split_images,
            "flux_variations": base_split_images * variation_count,
            "product_info_items": base_split_images * variation_count,
            "detail_pages": base_split_images * variation_count
        }
        
        # 检查实际完成数量
        def count_optimized_prompts(prompt_text):
            if not prompt_text:
                return 0
            lines = prompt_text.strip().split('\n')
            return len([line for line in lines if line.strip().startswith('-')])
        
        def count_list_items(data):
            if not data:
                return 0
            if isinstance(data, list):
                return len(data)
            return 0
        
        def count_product_info_items(product_info):
            if not product_info:
                return 0
            if isinstance(product_info, list):
                return len(product_info)
            elif isinstance(product_info, dict):
                # 如果是单个产品信息对象，计为1
                return 1
            return 0
        
        # 计算实际完成数量（在回填前的初始统计）
        actual_counts = {
            "optimized_prompts": count_optimized_prompts(result_data.get("optimized_prompt")),
            "mj_images": count_list_items(result_data.get("mj_image_urls") or result_data.get("mj_image_url")),
            "split_images": count_list_items(result_data.get("split_images")),
            "flux_variations": count_list_items(result_data.get("flux_variations")),
            "product_info_items": count_product_info_items(result_data.get("product_info")),
            "detail_pages": count_list_items(result_data.get("detail_pages"))
        }
        
        # === 新增：数据库回填逻辑（仅在 workflow_id 可用时）===
        try:
            if workflow_id:
                # 当这些计数为0时，尝试从数据库回填
                need_refetch_mj = (actual_counts["mj_images"] == 0)
                need_refetch_split = (actual_counts["split_images"] == 0)
                need_refetch_flux = (actual_counts["flux_variations"] == 0)
                
                if need_refetch_mj or need_refetch_split or need_refetch_flux:
                    from app.crud.image import image_crud
                    group_id = f'workflow_{workflow_id}'
                    images = image_crud.get_images_by_group(db=self.db, group_id=group_id, skip=0, limit=1000)
                    
                    mj_images_urls = []
                    split_images_list = []
                    flux_variations_list = []
                    
                    for img in images:
                        source = None
                        try:
                            import json as _json
                            if img.tencent_response:
                                tr = _json.loads(img.tencent_response)
                                source = tr.get("source")
                                # 附带索引（如果存在）
                                if source == "mj_split":
                                    split_idx = tr.get("split_index")
                                elif source == "flux_variation":
                                    src_idx = tr.get("source_image_index")
                                    var_idx = tr.get("variation_index")
                        except Exception:
                            source = None
                        
                        # 分类填充
                        if source == "mj_generation" and need_refetch_mj:
                            if img.cos_url:
                                mj_images_urls.append(img.cos_url)
                        elif source == "mj_split" and need_refetch_split:
                            entry = {"url": img.cos_url}
                            if getattr(img, "original_filename", None):
                                entry["file_name"] = img.original_filename
                            # 如果有 split_index 也带上（非必须）
                            try:
                                if img.tencent_response:
                                    tr = _json.loads(img.tencent_response)
                                    if tr.get("split_index") is not None:
                                        entry["split_index"] = tr.get("split_index")
                            except Exception:
                                pass
                            if img.cos_url:
                                split_images_list.append(entry)
                        elif source == "flux_variation" and need_refetch_flux:
                            entry = {"url": img.cos_url}
                            if getattr(img, "original_filename", None):
                                entry["file_name"] = img.original_filename
                            # 如果有索引信息也带上（非必须）
                            try:
                                if img.tencent_response:
                                    tr = _json.loads(img.tencent_response)
                                    if tr.get("source_image_index") is not None:
                                        entry["source_image_index"] = tr.get("source_image_index")
                                    if tr.get("variation_index") is not None:
                                        entry["variation_index"] = tr.get("variation_index")
                            except Exception:
                                pass
                            if img.cos_url:
                                flux_variations_list.append(entry)
                    
                    changed = False
                    if need_refetch_mj and mj_images_urls:
                        # 统一用数组字段
                        result_data["mj_image_urls"] = mj_images_urls
                        changed = True
                    if need_refetch_split and split_images_list:
                        result_data["split_images"] = split_images_list
                        changed = True
                    if need_refetch_flux and flux_variations_list:
                        result_data["flux_variations"] = flux_variations_list
                        changed = True
                    
                    # 如有回填，持久化保存，并重新计算 actual_counts
                    if changed:
                        ai_workflow.update_workflow_result(db=self.db, workflow_id=workflow_id, result_data=result_data)
                        
                        actual_counts = {
                            "optimized_prompts": count_optimized_prompts(result_data.get("optimized_prompt")),
                            "mj_images": count_list_items(result_data.get("mj_image_urls") or result_data.get("mj_image_url")),
                            "split_images": count_list_items(result_data.get("split_images")),
                            "flux_variations": count_list_items(result_data.get("flux_variations")),
                            "product_info_items": count_product_info_items(result_data.get("product_info")),
                            "detail_pages": count_list_items(result_data.get("detail_pages"))
                        }
        except Exception as _e:
            print(f"⚠️ 回填图片数据失败（不影响后续流程）：{_e}")
        # === 新增逻辑结束 ===
        
        # 判断步骤是否完成（实际数量 >= 预期数量）
        steps_status = {
            "step1_optimize_prompt": actual_counts["optimized_prompts"] >= expected_counts["optimized_prompts"],
            "step2_mj_generation": actual_counts["mj_images"] >= expected_counts["mj_images"],
            "step3_mj_split": actual_counts["split_images"] >= expected_counts["split_images"],
            "step4_flux_variation": actual_counts["flux_variations"] >= expected_counts["flux_variations"],
            "step5_generate_product_info": actual_counts["product_info_items"] >= expected_counts["product_info_items"],
            "step6_generate_detail_page": actual_counts["detail_pages"] >= expected_counts["detail_pages"]
        }
        
        # 打印详细的完成情况
        print(f"🔍 步骤完成情况详细分析:")
        print(f"🔍 variation_count: {variation_count}")
        for step, completed in steps_status.items():
            step_name = step.replace('step', '步骤').replace('_', ' ')
            if step == "step1_optimize_prompt":
                key = "optimized_prompts"
            elif step == "step2_mj_generation":
                key = "mj_images"
            elif step == "step3_mj_split":
                key = "split_images"
            elif step == "step4_flux_variation":
                key = "flux_variations"
            elif step == "step5_generate_product_info":
                key = "product_info_items"
            elif step == "step6_generate_detail_page":
                key = "detail_pages"
            
            status_icon = "✅" if completed else "❌"
            print(f"🔍 {status_icon} {step_name}: {actual_counts[key]}/{expected_counts[key]}")
        
        return steps_status
    
    async def _execute_from_step(self, workflow_id: int, steps_status: dict, result_data: dict) -> dict:
        """从指定步骤开始执行工作流"""
        from app.crud.ai_workflow import ai_workflow
        from app.models.ai_workflow import TaskStatus
        
        workflow = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
        try:
            # 步骤1: 优化提示词
            if not steps_status["step1_optimize_prompt"]:
                print(f"🔍 执行步骤1: 优化提示词")
                ai_workflow.update_workflow_status(
                    db=self.db, workflow_id=workflow_id,
                    status=TaskStatus.RUNNING, current_step="智能体优化提示词", progress=15
                )
                
                input_data = workflow.input_data or {}
                image_path = input_data.get('image_path') or input_data.get('image_url')
                # 新增：兼容批量任务的 image_urls
                if not image_path:
                    image_urls = input_data.get('image_urls') or []
                    if isinstance(image_urls, list) and image_urls:
                        image_path = image_urls[0]
                user_prompt = input_data.get('user_prompt', '')
                
                optimized_prompt = await self.step1_optimize_prompt(image_path, user_prompt, workflow_id)
                result_data["optimized_prompt"] = optimized_prompt
                
                ai_workflow.update_workflow_result(db=self.db, workflow_id=workflow_id, result_data=result_data)
                steps_status["step1_optimize_prompt"] = True
            else:
                print(f"✅ 步骤1已完成，跳过")
            
            # 步骤2: MJ文生图
            if not steps_status["step2_mj_generation"]:
                print(f"🔍 执行步骤2: MJ文生图")
                ai_workflow.update_workflow_status(
                    db=self.db, workflow_id=workflow_id,
                    status=TaskStatus.RUNNING, current_step="MJ文生图", progress=30
                )

                # 获取工作流信息，判断是否为批量处理
                workflow = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
                input_data = workflow.input_data or {}
                
                # 为批量处理工作流提供第一张图片的URL
                image_url = None
                if "image_urls" in input_data and input_data["image_urls"]:
                    image_url = input_data["image_urls"][0]  # 使用第一张图片
                elif "image_url" in input_data:
                    image_url = input_data["image_url"]
                elif "image_path" in input_data:
                    image_url = input_data["image_path"]
                
                # 传入 resume_from_existing=True 标志和图片URL
                mj_image_urls = await self.step2_mj_generation(
                    result_data["optimized_prompt"], 
                    workflow_id, 
                    image_url=image_url,  # 添加这个参数
                    resume_from_existing=True
                )
                result_data["mj_image_urls"] = mj_image_urls
                
                ai_workflow.update_workflow_result(db=self.db, workflow_id=workflow_id, result_data=result_data)
                steps_status["step2_mj_generation"] = True
            else:
                print(f"✅ 步骤2已完成，跳过")
                # 兼容旧数据格式
                if "mj_image_url" in result_data and "mj_image_urls" not in result_data:
                    result_data["mj_image_urls"] = [result_data["mj_image_url"]]
            
            # 步骤3: MJ图片拆分
            if not steps_status["step3_mj_split"]:
                print(f"🔍 执行步骤3: MJ图片拆分")
                ai_workflow.update_workflow_status(
                    db=self.db, workflow_id=workflow_id,
                    status=TaskStatus.RUNNING, current_step="MJ图片拆分", progress=45
                )
                
                all_split_images = []
                mj_image_urls = result_data.get("mj_image_urls", [])
                # 过滤掉None值和无效的URL
                valid_mj_image_urls = [url for url in mj_image_urls if url and isinstance(url, str) and url.strip()]
                print(f"🔍 有效的MJ图片URL数量: {len(valid_mj_image_urls)}/{len(mj_image_urls)}")
                
                for i, mj_image_url in enumerate(valid_mj_image_urls):
                    print(f"🔍 拆分第{i+1}张MJ图片: {mj_image_url}")
                    try:
                        split_images = await self.step3_mj_split(mj_image_url, workflow_id)
                        if split_images:
                            all_split_images.extend(split_images)
                            print(f"✅ 第{i+1}张MJ图片拆分成功")
                    except Exception as e:
                        print(f"❌ 第{i+1}张MJ图片拆分失败: {str(e)}")
                        continue
                
                if not all_split_images:
                    raise Exception("所有MJ图片拆分都失败了")
                
                result_data["split_images"] = all_split_images
                ai_workflow.update_workflow_result(db=self.db, workflow_id=workflow_id, result_data=result_data)
                steps_status["step3_mj_split"] = True
            else:
                print(f"✅ 步骤3已完成，跳过")
            
            # 步骤4: Flux裂变
            if not steps_status["step4_flux_variation"]:
                print(f"🔍 执行步骤4: Flux裂变")
                ai_workflow.update_workflow_status(
                    db=self.db, workflow_id=workflow_id,
                    status=TaskStatus.RUNNING, current_step="Flux图片裂变", progress=60
                )
                
                flux_variations = await self.step4_flux_variation(result_data["split_images"], workflow_id)
                result_data["flux_variations"] = flux_variations
                
                ai_workflow.update_workflow_result(db=self.db, workflow_id=workflow_id, result_data=result_data)
                steps_status["step4_flux_variation"] = True
            else:
                print(f"✅ 步骤4已完成，跳过")
            
            # 步骤5: 生成产品信息
            if not steps_status["step5_generate_product_info"]:
                print(f"🔍 执行步骤5: 生成产品信息")
                ai_workflow.update_workflow_status(
                    db=self.db, workflow_id=workflow_id,
                    status=TaskStatus.RUNNING, current_step="生成产品信息", progress=75
                )
                
                product_info = await self.step5_generate_product_info(result_data["optimized_prompt"], workflow_id, resume_from_existing=True)
                result_data["product_info"] = product_info
                
                ai_workflow.update_workflow_result(db=self.db, workflow_id=workflow_id, result_data=result_data)
                steps_status["step5_generate_product_info"] = True
            else:
                print(f"✅ 步骤5已完成，跳过")
            
            # 步骤6: 生成详情页
            if not steps_status["step6_generate_detail_page"]:
                print(f"🔍 执行步骤6: 生成详情页")
                ai_workflow.update_workflow_status(
                    db=self.db, workflow_id=workflow_id,
                    status=TaskStatus.RUNNING, current_step="生成详情页", progress=90
                )
                
                detail_pages = await self.step6_generate_detail_page(result_data["product_info"], result_data["flux_variations"], workflow_id, resume_from_existing=True)
                result_data["detail_pages"] = detail_pages
                
                ai_workflow.update_workflow_result(db=self.db, workflow_id=workflow_id, result_data=result_data)
                steps_status["step6_generate_detail_page"] = True
            else:
                print(f"✅ 步骤6已完成，跳过")
            
            # 完成工作流
            ai_workflow.update_workflow_status(
                db=self.db, workflow_id=workflow_id,
                status=TaskStatus.COMPLETED, current_step="处理完成", progress=100
            )
            
            print(f"✅ 工作流 {workflow_id} 执行完成")
            return result_data
            
        except Exception as e:
            error_msg = f"工作流执行失败: {str(e)}"
            print(f"❌ {error_msg}")
            
            ai_workflow.update_workflow_status(
                db=self.db, workflow_id=workflow_id,
                status=TaskStatus.FAILED, error_message=error_msg
            )
            
            raise Exception(error_msg)
    
    async def retry_single_step(self, workflow_id: int, step_name: str) -> dict:
        """重试单个步骤"""
        try:
            from app.crud.ai_workflow import ai_workflow
            
            workflow = ai_workflow.get_workflow(db=self.db, workflow_id=workflow_id)
            if not workflow:
                raise Exception(f"未找到工作流 ID: {workflow_id}")
            
            result_data = workflow.result_data or {}
            print(f"🔍 重试步骤: {step_name}")
            
            # === 新增：步骤名别名映射，增强兼容性 ===
            step_aliases = {
                "step1": "step1_optimize_prompt",
                "step2": "step2_mj_generation",
                "step3": "step3_mj_split",
                "step4": "step4_flux_variation",
                "step5": "step5_generate_product_info",
                "step6": "step6_generate_detail_page",
                "optimize_prompt": "step1_optimize_prompt",
                "mj_generation": "step2_mj_generation",
                "mj_split": "step3_mj_split",
                "flux_variation": "step4_flux_variation",
                "generate_product_info": "step5_generate_product_info",
                "generate_detail_page": "step6_generate_detail_page",
            }
            step_name = step_aliases.get(step_name, step_name)
            
            # === 新增：在执行步骤前，尽可能从数据库回填图片数据 ===
            try:
                # 仅当相关字段缺失时回填，避免覆盖已有内存状态
                need_mj = not (result_data.get("mj_image_urls") or result_data.get("mj_image_url"))
                need_split = not result_data.get("split_images")
                need_flux = not result_data.get("flux_variations")
                
                if need_mj or need_split or need_flux:
                    from app.crud.image import image_crud
                    import json as _json
                    group_id = f"workflow_{workflow_id}"
                    images = image_crud.get_images_by_group(db=self.db, group_id=group_id, skip=0, limit=1000)
                    
                    mj_urls, split_list, flux_list = [], [], []
                    for img in images:
                        try:
                            tr = _json.loads(img.tencent_response) if img.tencent_response else {}
                            source = tr.get("source")
                        except Exception:
                            source = None
                        
                        if source == "mj_generation" and need_mj and img.cos_url:
                            mj_urls.append(img.cos_url)
                        elif source == "mj_split" and need_split and img.cos_url:
                            entry = {"url": img.cos_url}
                            if getattr(img, "original_filename", None):
                                entry["file_name"] = img.original_filename
                            try:
                                if img.tencent_response:
                                    tr = _json.loads(img.tencent_response)
                                    if tr.get("split_index") is not None:
                                        entry["split_index"] = tr.get("split_index")
                            except Exception:
                                pass
                            split_list.append(entry)
                        elif source == "flux_variation" and need_flux and img.cos_url:
                            entry = {"url": img.cos_url}
                            if getattr(img, "original_filename", None):
                                entry["file_name"] = img.original_filename
                            try:
                                if img.tencent_response:
                                    tr = _json.loads(img.tencent_response)
                                    if tr.get("source_image_index") is not None:
                                        entry["source_image_index"] = tr.get("source_image_index")
                                    if tr.get("variation_index") is not None:
                                        entry["variation_index"] = tr.get("variation_index")
                            except Exception:
                                pass
                            flux_list.append(entry)
                    
                    changed = False
                    if need_mj and mj_urls:
                        result_data["mj_image_urls"] = mj_urls
                        changed = True
                    if need_split and split_list:
                        result_data["split_images"] = split_list
                        changed = True
                    if need_flux and flux_list:
                        result_data["flux_variations"] = flux_list
                        changed = True
                    
                    if changed:
                        ai_workflow.update_workflow_result(db=self.db, workflow_id=workflow_id, result_data=result_data)
                        print(f"🔁 已从DB回填图片数据：mj={len(result_data.get('mj_image_urls', []))}, split={len(result_data.get('split_images', []))}, flux={len(result_data.get('flux_variations', []))}")
            except Exception as _e:
                print(f"⚠️ 重试前回填图片数据失败（忽略继续）：{_e}")
            # === 回填结束 ===
            
            if step_name == "step1_optimize_prompt":
                    input_data = workflow.input_data or {}
                    image_path = input_data.get('image_path') or input_data.get('image_url')
                    # 新增：兼容批量任务的 image_urls
                    if not image_path:
                        image_urls = input_data.get('image_urls') or []
                        if isinstance(image_urls, list) and image_urls:
                            image_path = image_urls[0]
                    user_prompt = input_data.get('user_prompt', '')
                    
                    optimized_prompt = await self.step1_optimize_prompt(image_path, user_prompt, workflow_id)
                    result_data["optimized_prompt"] = optimized_prompt
                    
            elif step_name == "step2_mj_generation":
                if not result_data.get("optimized_prompt"):
                    raise Exception("缺少优化提示词，请先执行步骤1")
                
                # 新增：从 input_data 中获取图片路径（兼容批量与单图）
                input_data = workflow.input_data or {}
                image_url_param = None
                if "image_urls" in input_data and input_data["image_urls"]:
                    image_url_param = input_data["image_urls"][0]  # 批量任务使用第一张图
                elif "image_url" in input_data:
                    image_url_param = input_data["image_url"]
                elif "image_path" in input_data:
                    image_url_param = input_data["image_path"]
                
                # 传入 image_url 与 resume从existing，避免重复生成与“未找到图片路径”错误
                mj_image_urls = await self.step2_mj_generation(
                    result_data["optimized_prompt"],
                    workflow_id,
                    image_url=image_url_param,
                    resume_from_existing=True
                )
                result_data["mj_image_urls"] = mj_image_urls
                
            elif step_name == "step3_mj_split":
                if not result_data.get("mj_image_urls") and not result_data.get("mj_image_url"):
                    raise Exception("缺少MJ图片，请先执行步骤2")
                
                # 兼容旧数据格式
                mj_image_urls = result_data.get("mj_image_urls") or [result_data.get("mj_image_url")]
                all_split_images = []
                
                for i, mj_image_url in enumerate(mj_image_urls):
                    try:
                        split_images = await self.step3_mj_split(mj_image_url, workflow_id)
                        if split_images:
                            all_split_images.extend(split_images)
                    except Exception as e:
                        print(f"❌ 第{i+1}张图片拆分失败: {str(e)}")
                        continue
                
                if not all_split_images:
                    raise Exception("所有图片拆分都失败了")
                
                result_data["split_images"] = all_split_images
                
            elif step_name == "step4_flux_variation":
                if not result_data.get("split_images"):
                    raise Exception("缺少拆分图片，请先执行步骤3")
                
                flux_variations = await self.step4_flux_variation(result_data["split_images"], workflow_id)
                result_data["flux_variations"] = flux_variations
                
            elif step_name == "step5_generate_product_info":
                if not result_data.get("optimized_prompt"):
                    raise Exception("缺少优化提示词，请先执行步骤1")
                
                product_info = await self.step5_generate_product_info(result_data["optimized_prompt"], workflow_id)
                result_data["product_info"] = product_info
                
            elif step_name == "step6_generate_detail_page":
                if not result_data.get("flux_variations"):
                    raise Exception("缺少Flux变体图片，请先执行步骤4")
                
                # 如缺少product_info，尝试兜底生成
                if not result_data.get("product_info"):
                    if not result_data.get("optimized_prompt"):
                        raise Exception("缺少产品信息与优化提示词，请先执行步骤1或5")
                    product_info = await self.step5_generate_product_info(result_data["optimized_prompt"], workflow_id)
                    result_data["product_info"] = product_info
                
                # 统一保存为 detail_pages（与主流程一致）
                detail_pages = await self.step6_generate_detail_page(result_data["product_info"], result_data["flux_variations"], workflow_id)
                result_data["detail_pages"] = detail_pages
                
            else:
                raise Exception(f"未知的步骤名称: {step_name}")
            
            # 保存结果
            ai_workflow.update_workflow_result(db=self.db, workflow_id=workflow_id, result_data=result_data)
            
            print(f"✅ 步骤 {step_name} 重试成功")
            return result_data
            
        except Exception as e:
            error_msg = f"步骤 {step_name} 重试失败: {str(e)}"
            print(f"❌ {error_msg}")
            raise Exception(error_msg)